stack: MCP submit-Tool mit Upload Review Gate und Quarantäne-Schreibpfad (schliesst #32)

Files changed:
- .gitignore
- AGENTS.md
- CHANGES.md
- INSTALL-MCP.md
- README.md
- VERSION
- docs/why-gates-are-code.md
- instructions/gates.md
- instructions/ingest-queue.md
- instructions/mcp-read-server.md
- instructions/wiki-ingest/SKILL.md
- raw/CONTRACT.md
- tools/CONTRACT.md
- tools/chemenu/cli.py
- tools/chemenu/commands/docs_verify.py
- tools/chemenu/commands/doctor.py
- tools/chemenu/commands/upload_cmd.py
- tools/chemenu/config.py
- tools/chemenu/mcp/server.py
- tools/chemenu/tests/test_doctor.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

tools/chemenu/cli.py

@@ -31,6 +31,7 @@ try:
         search as search_module,
         touch as touch_module,
         types_cmd,
+        upload_cmd,
         upstream_cmd,
         version_cmd,
         work_cmd,
@@ -61,6 +62,7 @@ app.add_typer(index_build.app, name="index")
 app.add_typer(log_append.app, name="log")
 app.add_typer(provenance_cmd.app, name="sources")
 app.add_typer(raw_cmd.app, name="raw")
+app.add_typer(upload_cmd.app, name="upload")
 app.add_typer(instructions_cmd.app, name="instructions")
 app.add_typer(run_budget.app, name="budget")
 app.add_typer(types_cmd.app, name="types")

tools/chemenu/commands/docs_verify.py

@@ -120,6 +120,14 @@ REQUIRED_IGNORE_CANARIES = (
     # backstop a few lines above. Flat since Gitea #67 - incoming/ no longer
     # has type subdirectories, so the probe sits directly in it.
     "incoming/probe.pdf",
+    # 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",
+    # The `submit` tool's opt-in, same shape as `.wikitool-remotes.json`/
+    # `.wikitool-telemetry.json` a few lines below - per-checkout, never
+    # committed.
+    ".wikitool-upload.json",
 )
 REQUIRED_TRACKED_PATHS = (
     "reports/CONTRACT.md",

tools/chemenu/commands/doctor.py

@@ -376,6 +376,43 @@ def check_telemetry() -> Check:
     )
 
 
+def check_upload_intake() -> Check:
+    """Whether the MCP `submit` tool is armed for this checkout, and how full
+    its quarantine is.
+
+    Absent is the *safe* default here, unlike `check_publish_remotes`'s "any
+    push target passes" absence: no `.wikitool-upload.json` means the write
+    path does not exist at all, not that it is unrestricted - so this never
+    `FAIL`s on a missing file. 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.
+    """
+    from chemenu import upload as upload_module
+    from chemenu.errors import ValidationError
+
+    try:
+        cfg = upload_module.read_config(config.ROOT)
+    except ValidationError as exc:
+        return Check(
+            "upload-intake", "FAIL", str(exc),
+            f"Fix or delete {config.UPLOAD_CONFIG_FILENAME} - a broken one is not treated as "
+            "'no limits'",
+        )
+    if cfg is None:
+        return Check(
+            "upload-intake", "OK",
+            f"submit tool not registered - no {config.UPLOAD_CONFIG_FILENAME}",
+        )
+    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)}",
+    )
+
+
 def check_session_id() -> Check:
     import os
 
@@ -478,6 +515,7 @@ def run_doctor() -> list[Check]:
         check_conventions(),
         check_environment(),
         check_publish_remotes(),
+        check_upload_intake(),
         check_generated_files(),
         check_session_id(),
         check_telemetry(),
@@ -490,8 +528,8 @@ def doctor_command(
 ):
     """Check that this instance is correctly configured: dependencies, author,
     git identity/remote, published skills, structure, personalization, KB
-    conventions, generated files, session scoping, and telemetry state.
-    Read-only. Exits 1 only
+    conventions, generated files, session scoping, telemetry state, and
+    whether the MCP `submit` tool is armed. Read-only. Exits 1 only
     if a check FAILs."""
     checks = run_doctor()
 

tools/chemenu/commands/upload_cmd.py

@@ -0,0 +1,122 @@
+"""`wikitool upload list|show|accept|reject` - the human review side of the
+MCP submission quarantine (Gitea #32).
+
+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
+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`).
+"""
+from __future__ import annotations
+
+import json as json_module
+from typing import Optional
+
+import typer
+
+from chemenu import 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/.")
+
+
+def _call(fn, *args, **kwargs):
+    try:
+        return fn(*args, **kwargs)
+    except ValidationError as exc:
+        fail(str(exc))
+
+
+@app.command("list")
+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."""
+    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/.")
+        return
+    for manifest in manifests:
+        typer.echo(
+            f"{manifest['id']}  {manifest['filename']}  {manifest['size']}B  "
+            f"from {manifest['submitter']}"
+        )
+
+
+@app.command("show")
+def upload_show_command(
+    submission_id: str = typer.Argument(..., help="A submission id from `upload list`"),
+    json_out: bool = typer.Option(False, "--json"),
+):
+    """Print one submission's manifest in full - what a reviewer checks
+    before `accept`."""
+    manifest = _call(upload.read_manifest, config.ROOT, submission_id)
+    if json_out:
+        typer.echo(json_module.dumps(manifest, indent=2))
+        return
+    for key in ("id", "filename", "size", "sha256", "submitter", "submitter_source", "submitted_at"):
+        typer.echo(f"{key}: {manifest.get(key)}")
+
+
+def _clearance_message(manifest: dict, token: str, stale: Optional[str]) -> str:
+    lines = [
+        f"Upload Review Gate: submission '{manifest['id']}' needs a human to look at it "
+        "before it is promoted into incoming/.",
+        "",
+        f"  filename:         {manifest['filename']}",
+        f"  size:             {manifest['size']} bytes",
+        f"  sha256:           {manifest['sha256']}",
+        f"  submitter:        {manifest['submitter']}",
+        f"  submitter_source: {manifest['submitter_source']}",
+        f"  submitted_at:     {manifest['submitted_at']}",
+        "",
+    ]
+    if stale:
+        lines.append(
+            f"The token you passed ({stale}) does not match this submission - its manifest "
+            "changed, or the token was invented."
+        )
+        lines.append("")
+    lines.append(
+        "Nothing was promoted. Check this against raw/CONTRACT.md \"What does not belong "
+        "here\" and instructions/ingest-queue.md, then re-run with the token below:"
+    )
+    lines.append("")
+    lines.append(f"  tools/wikitool upload accept {manifest['id']} --confirm {token}")
+    return "\n".join(lines)
+
+
+@app.command("accept")
+def upload_accept_command(
+    submission_id: str = typer.Argument(..., help="A submission id from `upload list`"),
+    confirm: Optional[str] = typer.Option(
+        None, "--confirm", help="The token from a prior refusal, once a human has reviewed it"
+    ),
+):
+    """Promote a submission into incoming/ - refuses with Exit 42 until a
+    human has seen the manifest and cleared it with `--confirm <token>`."""
+    manifest = _call(upload.read_manifest, config.ROOT, submission_id)
+    token = upload.confirm_token(manifest)
+    if confirm != token:
+        needs_clearance(_clearance_message(manifest, token, confirm))
+        return
+    dest = _call(upload.promote, config.ROOT, submission_id)
+    success(f"Promoted '{submission_id}' to {rel_path(dest)}.")
+
+
+@app.command("reject")
+def upload_reject_command(
+    submission_id: str = typer.Argument(..., help="A submission id from `upload list`"),
+    reason: str = typer.Option(..., "--reason", help="Why this submission was declined"),
+):
+    """Delete a submission's material, keeping its ledger trail."""
+    _call(upload.reject, config.ROOT, submission_id, reason)
+    success(f"Rejected '{submission_id}': {reason}")

tools/chemenu/config.py

@@ -72,6 +72,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",),
     # 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).
@@ -236,6 +240,14 @@ TELEMETRY_FILENAME = ".wikitool-telemetry.json"
 # instructions/gates.md.
 PUBLISH_REMOTES_FILENAME = ".wikitool-remotes.json"
 
+# Opt-in for the MCP server's `submit` tool (Gitea #32): identity header name,
+# size deckel, extension allowlist, per-submitter quota. Same shape as the two
+# above - per-checkout, gitignored, no `.template` - but its absence means
+# something stronger than "unrestricted": **the write path does not exist at
+# all**, the tool is not registered. The safe direction, and a structural
+# opt-in rather than a flag - see `chemenu.upload.read_config`.
+UPLOAD_CONFIG_FILENAME = ".wikitool-upload.json"
+
 
 def default_author() -> str | None:
     """The author to stamp a new source page with, per instance.

tools/chemenu/mcp/server.py

@@ -1,17 +1,30 @@
-"""The MCP read server: `search`, `types`, `lint` and `status` over `kb/`.
+"""The MCP server: `search`, `types`, `lint`, `status` over `kb/`, plus an
+opt-in write path, `submit`, into a quarantine (Gitea #32).
 
 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
 at a terminal go through the same code, and a golden test holds the two
 outputs against each other rather than trusting that they agree.
 
-**There is no write path, structurally.** Nothing under `chemenu.commands` is
-imported here or in `chemenu.api`, so `new`, `touch`, `xref`, `cite`,
-`publish`, `migrate` and `version bump` are not reachable - the functions do
+**Five tools have no write path, structurally.** Nothing under
+`chemenu.commands` is imported here or in `chemenu.api`, so `new`, `touch`,
+`xref`, `cite`, `publish`, `migrate`, `version bump`, and the reviewer
+commands `upload accept`/`upload reject` are not reachable - the functions do
 not exist in this process's reach, rather than being filtered out of a list. A
 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 `.wikitool-upload.json` opts this checkout into it
+(`config.UPLOAD_CONFIG_FILENAME`, 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
+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`.
+
 **Authentication and rate limiting are not here.** Both are Traefik middleware
 in front of the process, per the operator's decision of 2026-09-01: a request
 that is not cleanly authenticated does not reach Python at all. What *is* here
@@ -37,10 +50,10 @@ import os
 from pathlib import Path
 from typing import Any, Optional
 
-from mcp.server.mcpserver import MCPServer
+from mcp.server.mcpserver import Context, MCPServer
 from mcp.server.mcpserver.exceptions import ToolError
 
-from chemenu import config
+from chemenu import config, upload
 from chemenu.api import Corpus
 from chemenu.errors import ChemenuError
 from chemenu.telemetry import policy
@@ -114,6 +127,20 @@ def build_server(
     if check_trace:
         check_trace_destination(corpus.root)
 
+    # A malformed `.wikitool-upload.json` is a start failure, not "no limits":
+    # this file 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 file would only surface on the first `submit`
+    # call. `upload_cfg` is None exactly when the tool below is not defined.
+    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 "
+        "reach kb/. A submission needs a human to promote it (see instructions/ingest-queue.md)."
+    )
     server = MCPServer(
         name=SERVER_NAME,
         instructions=(
@@ -121,7 +148,7 @@ def build_server(
             "Every answer carries the commit it was computed from ('commit') and "
             "when it was produced ('as_of'); a null commit means the served tree "
             "has uncommitted changes and the answer corresponds to no revision. "
-            "This server is read-only - there is no tool that writes."
+            f"{read_only_note}"
         ),
     )
 
@@ -206,6 +233,34 @@ def build_server(
     def status() -> dict[str, Any]:
         return _guard(corpus.status)
 
+    if upload_cfg is not None:
+
+        @server.tool(
+            name="submit",
+            description=(
+                "Submit a document into review, not into the wiki. The file is base64-"
+                "encoded and written into a quarantine no other tool or command reads - "
+                "a human reviews and promotes it later (wikitool upload accept), or "
+                "rejects it. This never writes to kb/, directly or indirectly. Refused "
+                f"without an identity on the request. Limits: up to {upload_cfg.max_bytes} "
+                f"bytes, extensions {', '.join(upload_cfg.allowed_extensions)}, "
+                f"{upload_cfg.submissions_per_day} submission(s) and "
+                f"{upload_cfg.bytes_per_day} byte(s) per submitter per rolling 24h."
+            ),
+        )
+        def submit(filename: str, content_base64: str, ctx: Context) -> dict[str, Any]:
+            """`filename` is a bare name, never a path. `content_base64` is the
+            whole file, base64-encoded. The submitter's identity comes from
+            this request's own headers (`ctx.headers`), never from an
+            argument - a caller cannot claim to be someone else by passing a
+            different value here, because there is no such value to pass."""
+            headers = ctx.headers or {}
+            submitter = headers.get(upload_cfg.identity_header)
+            return _guard(lambda: upload.submit(
+                corpus.root, upload_cfg,
+                filename=filename, content_b64=content_base64, submitter=submitter,
+            ))
+
     return server
 
 

tools/chemenu/tests/test_doctor.py

@@ -79,6 +79,10 @@ def _status(checks, name):
     return next(c.status for c in checks if c.name == name)
 
 
+def _detail(checks, name) -> str:
+    return next(c.detail for c in checks if c.name == name)
+
+
 def test_healthy_instance_has_no_fail(instance):
     checks = doctor.run_doctor()
     assert not any(c.status == "FAIL" for c in checks)
@@ -317,6 +321,31 @@ def test_publish_remotes_warns_on_several_remotes_without_an_allowlist(instance)
     assert "not armed" in _remotes_detail(checks)
 
 
+def test_upload_intake_ok_when_no_config_file(instance):
+    checks = doctor.run_doctor()
+    assert _status(checks, "upload-intake") == "OK"
+    assert "not registered" in _detail(checks, "upload-intake")
+
+
+def test_upload_intake_ok_and_armed_when_config_is_valid(instance):
+    (config.ROOT / config.UPLOAD_CONFIG_FILENAME).write_text(
+        '{"schema": 1, "max_bytes": 100, "allowed_extensions": [".md"], '
+        '"quota": {"submissions_per_day": 1, "bytes_per_day": 100}}',
+        encoding="utf-8",
+    )
+    checks = doctor.run_doctor()
+    assert _status(checks, "upload-intake") == "OK"
+    detail = _detail(checks, "upload-intake")
+    assert "armed" in detail
+    assert "0 submission(s) waiting" in detail
+
+
+def test_upload_intake_fails_on_a_malformed_config(instance):
+    (config.ROOT / config.UPLOAD_CONFIG_FILENAME).write_text("{not json", encoding="utf-8")
+    checks = doctor.run_doctor()
+    assert _status(checks, "upload-intake") == "FAIL"
+
+
 def test_missing_generated_file_fails(instance):
     config.LOG_FILE.unlink()
     checks = doctor.run_doctor()

tools/chemenu/tests/test_mcp_server.py

@@ -1,18 +1,22 @@
-"""Tests for the MCP read server (Gitea #19).
+"""Tests for the MCP server (Gitea #19, plus `submit` from #32).
 
 The acceptance criteria that are not about a value coming back:
 
 - the wire format **is** the CLI's `--json` form, held together by a golden
   test rather than by intention;
-- no path of the server writes into `kb/`, `reports/` or git;
-- there is no write tool, because nothing under `chemenu.commands` is
-  importable from it - structural, not filtered;
+- no path of the five original tools writes into `kb/`, `reports/` or git;
+- five tools have no write path, because nothing under `chemenu.commands` is
+  importable from the server - structural, not filtered;
 - every response carries the commit it was computed from;
-- telemetry cannot land inside the served checkout.
+- telemetry cannot land inside the served checkout;
+- the sixth tool, `submit`, exists only when `.wikitool-upload.json` opts a
+  checkout in, writes only under `mcp-upload/`, and never resolves an
+  identity from anywhere but the request's own headers.
 """
 from __future__ import annotations
 
 import asyncio
+import base64
 import json
 import shutil
 import subprocess
@@ -21,11 +25,15 @@ from pathlib import Path
 
 import pytest
 
-from chemenu import config
+from chemenu import config, upload
+from chemenu.errors import ValidationError
 
 pytest.importorskip("mcp", reason="the MCP server's dependency is optional; see "
                                   "tools/requirements-mcp.txt")
 
+from mcp.server.context import ServerRequestContext  # noqa: E402
+from mcp.server.mcpserver import Context as ToolContext  # noqa: E402
+
 from chemenu.mcp.server import (  # noqa: E402  - after the skip guard
     TRANSPORTS,
     TraceWouldWriteIntoCorpus,
@@ -34,6 +42,31 @@ from chemenu.mcp.server import (  # noqa: E402  - after the skip guard
     serve,
 )
 
+_UPLOAD_CFG = {
+    "schema": 1,
+    "identity_header": "X-Forwarded-User",
+    "max_bytes": 1024,
+    "allowed_extensions": [".md"],
+    "quota": {"submissions_per_day": 10, "bytes_per_day": 100_000},
+}
+
+
+class _FakeRequest:
+    def __init__(self, headers):
+        self.headers = headers
+
+
+def _context(server, headers: dict | None = None) -> ToolContext:
+    """A `Context` carrying `headers` the way a real HTTP transport would -
+    `None` simulates stdio, where the SDK's own docstring says `ctx.headers`
+    is `None`."""
+    request = _FakeRequest(headers) if headers is not None else None
+    request_context = ServerRequestContext(
+        session=None, lifespan_context={}, protocol_version="2025-06-18",
+        method="tools/call", request=request,
+    )
+    return ToolContext(mcp_server=server, request_context=request_context)
+
 
 @pytest.fixture
 def corpus(tmp_path: Path) -> Path:
@@ -291,3 +324,96 @@ def test_a_refused_trace_destination_stops_the_process_with_a_message(monkeypatc
     monkeypatch.delenv("WIKI_TRACE_DIR", raising=False)
     assert entry.main([]) == 1
     assert "WIKI_TRACE" in capsys.readouterr().err
+
+
+# --- submit (Gitea #32) --------------------------------------------------------
+
+def _arm_upload(corpus: Path) -> None:
+    (corpus / ".wikitool-upload.json").write_text(json.dumps(_UPLOAD_CFG), encoding="utf-8")
+    (corpus / "incoming").mkdir(exist_ok=True)
+
+
+def test_submit_is_absent_without_the_upload_config(corpus):
+    server = build_server(corpus, check_trace=False)
+    names = {tool.name for tool in asyncio.run(server.list_tools())}
+    assert "submit" not in names
+
+
+def test_submit_is_present_once_armed(corpus):
+    _arm_upload(corpus)
+    server = build_server(corpus, check_trace=False)
+    names = {tool.name for tool in asyncio.run(server.list_tools())}
+    assert names == {"search", "types", "describe_type", "lint", "status", "submit"}
+
+
+def test_a_malformed_upload_config_refuses_to_build(corpus):
+    (corpus / ".wikitool-upload.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):
+    _arm_upload(corpus)
+    server = build_server(corpus, check_trace=False)
+    before = _tree(corpus)
+    content_b64 = base64.b64encode(b"a submitted file\n").decode("ascii")
+
+    ctx = _context(server, {"X-Forwarded-User": "alice"})
+    result = asyncio.run(server.call_tool(
+        "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
+    ))
+    if isinstance(result, tuple):
+        result = result[1]
+    assert not result.is_error, result.content
+    manifest = result.structured_content
+
+    assert manifest["submitter"] == "alice"
+    assert manifest["submitter_source"] == "X-Forwarded-User"
+
+    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 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
+    # 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.
+    status = subprocess.run(
+        ["git", "status", "--porcelain"], cwd=corpus, capture_output=True, text=True, check=True
+    ).stdout
+    changed_tracked = {line[3:] for line in status.splitlines() if not line.startswith("??")}
+    assert changed_tracked == set()
+
+
+def test_submit_without_identity_header_is_a_tool_error(corpus):
+    from mcp.server.mcpserver.exceptions import ToolError
+
+    _arm_upload(corpus)
+    server = build_server(corpus, check_trace=False)
+    content_b64 = base64.b64encode(b"data\n").decode("ascii")
+
+    ctx = _context(server, {})  # headers present, but not the configured one
+    with pytest.raises(ToolError):
+        asyncio.run(server.call_tool(
+            "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
+        ))
+
+
+def test_submit_on_stdio_with_no_headers_at_all_is_a_tool_error(corpus):
+    """`ctx.headers` is `None` on stdio, per the SDK's own docstring - the
+    tool must treat that the same as an absent identity header, not crash on
+    a `None.get(...)`."""
+    from mcp.server.mcpserver.exceptions import ToolError
+
+    _arm_upload(corpus)
+    server = build_server(corpus, check_trace=False)
+    content_b64 = base64.b64encode(b"data\n").decode("ascii")
+
+    ctx = _context(server, headers=None)
+    with pytest.raises(ToolError):
+        asyncio.run(server.call_tool(
+            "submit", {"filename": "note.md", "content_base64": content_b64}, context=ctx,
+        ))

tools/chemenu/tests/test_upload.py

@@ -0,0 +1,426 @@
+"""Tests for the MCP upload quarantine (Gitea #32) - `chemenu/upload.py`.
+
+`upload.py` takes `root` explicitly and never touches `config.ROOT`, so these
+tests point straight at `tmp_path` without repointing global state the way
+`kb_dir`/`raw_dir` do.
+"""
+from __future__ import annotations
+
+import base64
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from chemenu import upload
+from chemenu.errors import ValidationError
+
+CONTENT = b"hello world, this is a test submission\n"
+CONTENT_B64 = base64.b64encode(CONTENT).decode("ascii")
+
+
+def _cfg(**overrides) -> upload.UploadConfig:
+    defaults = dict(
+        identity_header="X-Forwarded-User",
+        max_bytes=1024,
+        allowed_extensions=(".md", ".pdf"),
+        submissions_per_day=5,
+        bytes_per_day=10_000,
+    )
+    defaults.update(overrides)
+    return upload.UploadConfig(**defaults)
+
+
+def _tree(root: Path) -> dict[str, bytes]:
+    return {
+        str(p.relative_to(root)): p.read_bytes()
+        for p in sorted(root.rglob("*"))
+        if p.is_file()
+    }
+
+
+# --- sanitize_filename ------------------------------------------------------
+
+@pytest.mark.parametrize("name", ["a/b.md", "a\\b.md", "..", ".", "", "   ", ".hidden",
+                                    "a\x00b.md"])
+def test_sanitize_filename_rejects_unsafe_names(name):
+    with pytest.raises(ValidationError):
+        upload.sanitize_filename(name)
+
+
+def test_sanitize_filename_accepts_a_bare_name():
+    assert upload.sanitize_filename("report.md") == "report.md"
+
+
+# --- read_config -------------------------------------------------------------
+
+def test_read_config_absent_is_none(tmp_path):
+    assert upload.read_config(tmp_path) is None
+
+
+def test_read_config_malformed_json_is_an_error_not_unrestricted(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text("{not json", encoding="utf-8")
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+def test_read_config_missing_field_is_an_error(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({"schema": 1, "max_bytes": 100}), encoding="utf-8"
+    )
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+def test_read_config_extension_without_dot_is_rejected(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({
+            "schema": 1, "max_bytes": 100, "allowed_extensions": ["md"],
+            "quota": {"submissions_per_day": 1, "bytes_per_day": 1},
+        }),
+        encoding="utf-8",
+    )
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+def test_read_config_rejects_a_non_object_json(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text("[1, 2, 3]", encoding="utf-8")
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+@pytest.mark.parametrize("value", [-1, 0])
+def test_read_config_rejects_a_non_positive_max_bytes(tmp_path, value):
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({
+            "schema": 1, "max_bytes": value, "allowed_extensions": [".md"],
+            "quota": {"submissions_per_day": 1, "bytes_per_day": 1},
+        }),
+        encoding="utf-8",
+    )
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+@pytest.mark.parametrize("key,value", [("submissions_per_day", 0), ("bytes_per_day", -5)])
+def test_read_config_rejects_non_positive_quota_values(tmp_path, key, value):
+    quota = {"submissions_per_day": 1, "bytes_per_day": 1}
+    quota[key] = value
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({"schema": 1, "max_bytes": 100, "allowed_extensions": [".md"], "quota": quota}),
+        encoding="utf-8",
+    )
+    with pytest.raises(ValidationError):
+        upload.read_config(tmp_path)
+
+
+def test_read_config_valid_file_parses(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({
+            "schema": 1, "identity_header": "X-User", "max_bytes": 2048,
+            "allowed_extensions": [".MD", ".pdf"],
+            "quota": {"submissions_per_day": 3, "bytes_per_day": 999},
+        }),
+        encoding="utf-8",
+    )
+    cfg = upload.read_config(tmp_path)
+    assert cfg.identity_header == "X-User"
+    assert cfg.max_bytes == 2048
+    assert cfg.allowed_extensions == (".md", ".pdf")
+    assert cfg.submissions_per_day == 3
+    assert cfg.bytes_per_day == 999
+
+
+def test_read_config_default_identity_header(tmp_path):
+    (tmp_path / ".wikitool-upload.json").write_text(
+        json.dumps({
+            "schema": 1, "max_bytes": 10,
+            "allowed_extensions": [".md"],
+            "quota": {"submissions_per_day": 1, "bytes_per_day": 10},
+        }),
+        encoding="utf-8",
+    )
+    assert upload.read_config(tmp_path).identity_header == upload.DEFAULT_IDENTITY_HEADER
+
+
+# --- 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()
+
+
+def test_write_primitive_refuses_an_absolute_path(tmp_path):
+    with pytest.raises(ValidationError):
+        upload._write_atomic_within(tmp_path, Path("/etc/passwd"), b"x")
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need admin on Windows")
+def test_write_primitive_refuses_a_symlink_escape(tmp_path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    upload_dir = tmp_path / "mcp-upload"
+    upload_dir.mkdir()
+    (upload_dir / "escape").symlink_to(outside, target_is_directory=True)
+
+    with pytest.raises(ValidationError):
+        upload._write_atomic_within(tmp_path, Path("escape") / "pwned.txt", b"x")
+    assert not (outside / "pwned.txt").exists()
+
+
+def test_write_primitive_writes_inside_and_is_atomic(tmp_path):
+    dest = upload._write_atomic_within(tmp_path, Path("sub") / "file.txt", b"payload")
+    assert dest.read_bytes() == b"payload"
+    # no leftover temp file
+    assert list(dest.parent.glob(".*.tmp")) == []
+
+
+# --- submit --------------------------------------------------------------------
+
+def test_submit_happy_path_writes_manifest_and_content(tmp_path):
+    manifest = upload.submit(
+        tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter="torben"
+    )
+    assert manifest["filename"] == "report.md"
+    assert manifest["submitter"] == "torben"
+    assert manifest["submitter_source"] == "X-Forwarded-User"
+    assert manifest["size"] == len(CONTENT)
+
+    submission_dir = tmp_path / "mcp-upload" / 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()
+    assert len(ledger_lines) == 1
+    event = json.loads(ledger_lines[0])
+    assert event["event"] == "submitted"
+    assert event["submitter"] == "torben"
+
+
+def test_submit_without_identity_writes_nothing(tmp_path):
+    before = _tree(tmp_path)
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter=None)
+    assert _tree(tmp_path) == before
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter="  ")
+    assert _tree(tmp_path) == before
+
+
+def test_submit_rejects_disallowed_extension(tmp_path):
+    before = _tree(tmp_path)
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(), filename="evil.exe", content_b64=CONTENT_B64, submitter="t")
+    assert _tree(tmp_path) == before
+
+
+def test_submit_rejects_oversized_before_decoding(tmp_path, monkeypatch):
+    """The base64-length precheck fires before base64.b64decode is even
+    called - simulated by making decode raise if it is ever reached."""
+    import chemenu.upload as upload_module
+
+    def _boom(*args, **kwargs):
+        raise AssertionError("decode should not run past the length precheck")
+
+    monkeypatch.setattr(upload_module.base64, "b64decode", _boom)
+    before = _tree(tmp_path)
+    huge_b64 = base64.b64encode(b"x" * 10_000).decode("ascii")
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(max_bytes=10), filename="a.md", content_b64=huge_b64, submitter="t")
+    assert _tree(tmp_path) == before
+
+
+def test_submit_rejects_oversized_by_a_wide_margin(tmp_path):
+    """`max_bytes` far below the real payload size - whichever of the two
+    checks (base64-length precheck, post-decode check) fires, nothing is
+    written."""
+    cfg = _cfg(max_bytes=len(CONTENT) - 10)
+    before = _tree(tmp_path)
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    assert _tree(tmp_path) == before
+
+
+def test_submit_does_not_reject_a_payload_sitting_exactly_at_the_limit(tmp_path):
+    """The base64-length precheck overestimates the decoded size by up to the
+    2 padding characters a valid encoding carries - it must not turn that
+    slack into a false rejection of a payload that is genuinely within
+    `max_bytes`."""
+    manifest = upload.submit(
+        tmp_path, _cfg(max_bytes=len(CONTENT)), filename="a.md",
+        content_b64=CONTENT_B64, submitter="t",
+    )
+    assert manifest["size"] == len(CONTENT)
+
+
+def test_submit_rejects_one_byte_over_the_limit(tmp_path):
+    """One byte over is still refused - whichever of the two checks catches
+    it, the boundary itself is exact."""
+    before = _tree(tmp_path)
+    with pytest.raises(ValidationError):
+        upload.submit(
+            tmp_path, _cfg(max_bytes=len(CONTENT) - 1), filename="a.md",
+            content_b64=CONTENT_B64, submitter="t",
+        )
+    assert _tree(tmp_path) == before
+
+
+def test_submit_rejects_invalid_base64(tmp_path):
+    before = _tree(tmp_path)
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(), filename="a.md", content_b64="not base64!!", submitter="t")
+    assert _tree(tmp_path) == before
+
+
+def test_submit_rejects_empty_payload(tmp_path):
+    with pytest.raises(ValidationError):
+        upload.submit(tmp_path, _cfg(), filename="a.md", content_b64="", submitter="t")
+
+
+def test_submit_rejects_a_duplicate_pending_hash_naming_the_waiting_id(tmp_path):
+    first = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    with pytest.raises(ValidationError, match=first["id"]):
+        upload.submit(tmp_path, _cfg(), filename="b.md", content_b64=CONTENT_B64, submitter="t")
+
+
+def test_submit_allows_resubmission_after_rejection(tmp_path):
+    first = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    upload.reject(tmp_path, first["id"], "not needed")
+    second = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    assert second["id"] != first["id"]
+
+
+def test_submit_enforces_submissions_per_day_quota(tmp_path):
+    cfg = _cfg(submissions_per_day=1)
+    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    other_content = base64.b64encode(b"different content here\n").decode("ascii")
+    with pytest.raises(ValidationError, match="Quota"):
+        upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="t")
+
+
+def test_submit_enforces_bytes_per_day_quota(tmp_path):
+    cfg = _cfg(bytes_per_day=len(CONTENT))
+    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    other_content = base64.b64encode(b"more bytes than allowed now\n").decode("ascii")
+    with pytest.raises(ValidationError, match="Quota"):
+        upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="t")
+
+
+def test_submit_quota_is_per_submitter(tmp_path):
+    cfg = _cfg(submissions_per_day=1)
+    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="alice")
+    other_content = base64.b64encode(b"different content again\n").decode("ascii")
+    # bob has his own quota - not blocked by alice's submission
+    upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="bob")
+
+
+# --- list / read ----------------------------------------------------------------
+
+def test_list_submissions_oldest_first(tmp_path):
+    a = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    other = base64.b64encode(b"second file content\n").decode("ascii")
+    b = upload.submit(tmp_path, _cfg(), filename="b.md", content_b64=other, submitter="t")
+    ids = [m["id"] for m in upload.list_submissions(tmp_path)]
+    assert ids == sorted([a["id"], b["id"]])
+
+
+def test_read_manifest_unknown_id_is_an_error(tmp_path):
+    with pytest.raises(ValidationError):
+        upload.read_manifest(tmp_path, "no-such-id")
+
+
+def test_read_manifest_path_traversal_id_is_rejected(tmp_path):
+    with pytest.raises(ValidationError):
+        upload.read_manifest(tmp_path, "../../etc/passwd")
+
+
+# --- confirm_token ----------------------------------------------------------------
+
+def test_confirm_token_is_deterministic_and_moves_with_content(tmp_path):
+    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    token1 = upload.confirm_token(manifest)
+    token2 = upload.confirm_token(dict(manifest))
+    assert token1 == token2
+    mutated = dict(manifest, size=manifest["size"] + 1)
+    assert upload.confirm_token(mutated) != token1
+
+
+# --- promote ------------------------------------------------------------------
+
+def test_promote_moves_file_deletes_dir_and_ledgers_accepted(tmp_path):
+    (tmp_path / "incoming").mkdir()
+    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    dest = upload.promote(tmp_path, manifest["id"])
+
+    assert dest == tmp_path / "incoming" / "a.md"
+    assert dest.read_bytes() == CONTENT
+    assert not (tmp_path / "mcp-upload" / manifest["id"]).exists()
+
+    events = [json.loads(line) for line in
+              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
+    assert [e["event"] for e in events] == ["submitted", "accepted"]
+
+    # nothing else in the tree changed
+    for stray in ("raw", "kb", "work", "reports"):
+        assert not (tmp_path / stray).exists()
+
+
+def test_promote_unknown_id_is_rejected(tmp_path):
+    with pytest.raises(ValidationError):
+        upload.promote(tmp_path, "no-such-id")
+
+
+def test_promote_refuses_when_the_file_is_missing_from_disk(tmp_path):
+    """The manifest exists but the content file was removed underneath it -
+    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()
+
+    with pytest.raises(ValidationError):
+        upload.promote(tmp_path, manifest["id"])
+
+
+def test_promote_refuses_when_incoming_already_has_the_name(tmp_path):
+    (tmp_path / "incoming").mkdir()
+    (tmp_path / "incoming" / "a.md").write_text("already here", encoding="utf-8")
+    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+
+    with pytest.raises(ValidationError):
+        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 / "incoming" / "a.md").read_text() == "already here"
+
+
+# --- reject -------------------------------------------------------------------
+
+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()
+    events = [json.loads(line) for line in
+              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
+    assert events[-1]["event"] == "rejected"
+    assert events[-1]["reason"] == "license unclear"
+    assert events[-1]["sha256"] == manifest["sha256"]
+
+
+def test_reject_requires_a_reason(tmp_path):
+    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
+    with pytest.raises(ValidationError):
+        upload.reject(tmp_path, manifest["id"], "")
+    # nothing deleted
+    assert (tmp_path / "mcp-upload" / manifest["id"]).exists()
+
+
+def test_reject_unknown_id_is_rejected(tmp_path):
+    with pytest.raises(ValidationError):
+        upload.reject(tmp_path, "no-such-id", "reason")

tools/chemenu/tests/test_upload_cmd.py

@@ -0,0 +1,109 @@
+"""Tests for `wikitool upload ...` (Gitea #32) - the Upload Review Gate and
+the two read commands a reviewer uses before clearing it."""
+from __future__ import annotations
+
+import base64
+import json
+
+import pytest
+import typer
+
+from chemenu import config, upload
+from chemenu.commands._util import EXIT_NEEDS_CLEARANCE
+from chemenu.commands.upload_cmd import (
+    upload_accept_command,
+    upload_list_command,
+    upload_reject_command,
+    upload_show_command,
+)
+
+CONTENT = b"a test submission for the CLI layer\n"
+CONTENT_B64 = base64.b64encode(CONTENT).decode("ascii")
+
+_CFG = upload.UploadConfig(
+    identity_header="X-Forwarded-User",
+    max_bytes=1024,
+    allowed_extensions=(".md",),
+    submissions_per_day=10,
+    bytes_per_day=100_000,
+)
+
+
+@pytest.fixture
+def root(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, "ROOT", tmp_path)
+    (tmp_path / "incoming").mkdir()
+    return tmp_path
+
+
+def _seed(root, filename="report.md", submitter="torben"):
+    return upload.submit(root, _CFG, filename=filename, content_b64=CONTENT_B64, submitter=submitter)
+
+
+def test_accept_without_a_token_needs_clearance(root, capsys):
+    manifest = _seed(root)
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_accept_command(submission_id=manifest["id"], confirm=None)
+    assert excinfo.value.exit_code == EXIT_NEEDS_CLEARANCE
+    out = capsys.readouterr().out
+    assert manifest["id"] in out
+    assert manifest["sha256"] in out
+    assert "--confirm" in out
+    # nothing moved
+    assert (root / "mcp-upload" / manifest["id"]).exists()
+    assert not (root / "incoming" / manifest["filename"]).exists()
+
+
+def test_accept_with_a_stale_token_needs_clearance_again(root):
+    manifest = _seed(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()
+
+
+def test_accept_with_the_right_token_promotes(root):
+    manifest = _seed(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()
+
+
+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()
+
+
+def test_reject_without_a_reason_is_a_validation_error(root):
+    manifest = _seed(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()
+
+
+def test_list_json_matches_pending_manifests(root, capsys):
+    manifest = _seed(root)
+    upload_list_command(json_out=True)
+    printed = json.loads(capsys.readouterr().out)
+    assert printed == [manifest]
+
+
+def test_list_reports_nothing_waiting(root, capsys):
+    upload_list_command(json_out=False)
+    assert "Nothing is waiting" in capsys.readouterr().out
+
+
+def test_show_unknown_id_is_a_validation_error(root):
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_show_command(submission_id="no-such-id", json_out=False)
+    assert excinfo.value.exit_code == 1
+
+
+def test_show_prints_the_manifest(root, capsys):
+    manifest = _seed(root)
+    upload_show_command(submission_id=manifest["id"], json_out=True)
+    printed = json.loads(capsys.readouterr().out)
+    assert printed == manifest

tools/chemenu/upload.py

@@ -0,0 +1,498 @@
+"""The MCP write path's one choke point: `mcp-upload/`, and nothing else
+(Gitea #32).
+
+`#19`'s MCP read server has an absence property - nothing under
+`chemenu.commands` is importable from it, so a write function does not exist
+in that process's reach. Once a `submit` tool exists that property stops
+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/`
+    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`),
+never a caller-supplied name, so there is no path to sanitise in the first
+place.
+
+Two stages, two different grants of trust:
+
+    mcp-upload/<id>/   material nobody has looked at - `submit` 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.
+
+Stdlib only, like `chemenu.telemetry` - this module is imported by the MCP
+server process on every request, not only at CLI dispatch.
+"""
+from __future__ import annotations
+
+import base64
+import binascii
+import datetime
+import hashlib
+import json
+import os
+import secrets
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Optional
+
+from chemenu import config
+from chemenu.errors import ValidationError
+
+DEFAULT_IDENTITY_HEADER = "X-Forwarded-User"
+
+# Submissions older than this stop counting toward a submitter's quota. A
+# rolling window rather than a calendar day - "resets at midnight" is a
+# surprise no operator asked for, and a rolling window needs nothing stored
+# beyond the ledger that already exists for other reasons.
+_QUOTA_WINDOW = datetime.timedelta(days=1)
+
+_FORBIDDEN_FILENAME_CHARS = ("/", "\\", "\x00")
+
+
+@dataclass(frozen=True)
+class UploadConfig:
+    """The opt-in, read from `.wikitool-upload.json` - see
+    `config.UPLOAD_CONFIG_FILENAME`."""
+
+    identity_header: str
+    max_bytes: int
+    allowed_extensions: tuple[str, ...]
+    submissions_per_day: int
+    bytes_per_day: int
+
+
+def read_config(root: "Path | str") -> Optional[UploadConfig]:
+    """The upload opt-in for `root`, or `None` when it is absent - which means
+    the write path does not exist, not that it is unrestricted.
+
+    A malformed file is a `ValidationError`, never a silent "no limits": this
+    file decides whether an unauthenticated write path is offered at all, so a
+    corrupted safeguard must not read as a disabled one - the same posture
+    `git_publish.read_allowed_push_urls` takes for `.wikitool-remotes.json`,
+    deliberately not the "ignore what does not parse" posture
+    `telemetry.policy` takes for its own config, because that one only ever
+    narrows an existing on/off default and this one creates a capability that
+    otherwise does not exist.
+    """
+    path = Path(root) / config.UPLOAD_CONFIG_FILENAME
+    if not path.is_file():
+        return None
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValidationError(
+            f"{config.UPLOAD_CONFIG_FILENAME} is unreadable ({exc}). It decides whether the "
+            "upload tool is offered at all, so a broken file is not treated as 'no limits' - "
+            "fix it or delete it deliberately."
+        ) from exc
+    if not isinstance(data, dict):
+        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME} must contain a JSON object.")
+
+    expected = (
+        '{"schema": 1, "identity_header": "X-Forwarded-User", "max_bytes": 10485760, '
+        '"allowed_extensions": [".md", ".pdf"], '
+        '"quota": {"submissions_per_day": 20, "bytes_per_day": 52428800}}'
+    )
+    try:
+        identity_header = str(data.get("identity_header") or DEFAULT_IDENTITY_HEADER)
+        max_bytes = int(data["max_bytes"])
+        raw_extensions = data["allowed_extensions"]
+        extensions = tuple(sorted({str(ext).lower() for ext in raw_extensions}))
+        quota = data["quota"]
+        submissions_per_day = int(quota["submissions_per_day"])
+        bytes_per_day = int(quota["bytes_per_day"])
+    except (KeyError, TypeError, ValueError) as exc:
+        raise ValidationError(
+            f"{config.UPLOAD_CONFIG_FILENAME} is missing or misshapes a required field ({exc}). "
+            f"Expected: {expected}"
+        ) from exc
+
+    if max_bytes <= 0:
+        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: max_bytes must be positive.")
+    if submissions_per_day <= 0 or bytes_per_day <= 0:
+        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: quota values must be positive.")
+    if not extensions:
+        raise ValidationError(f"{config.UPLOAD_CONFIG_FILENAME}: allowed_extensions must not be empty.")
+    not_dotted = [ext for ext in extensions if not ext.startswith(".")]
+    if not_dotted:
+        raise ValidationError(
+            f"{config.UPLOAD_CONFIG_FILENAME}: allowed_extensions must each start with '.': {not_dotted}"
+        )
+
+    return UploadConfig(
+        identity_header=identity_header,
+        max_bytes=max_bytes,
+        allowed_extensions=extensions,
+        submissions_per_day=submissions_per_day,
+        bytes_per_day=bytes_per_day,
+    )
+
+
+def sanitize_filename(name: str) -> str:
+    """A bare, safe basename, or `ValidationError` - never a path.
+
+    An einreicher chooses this string, so it is adversarial input: no path
+    separator, no `..`, no null byte, no leading dot (a dotfile is never a
+    legitimate submission name), no empty stem. What survives is still just a
+    name - the submission id (generated, never caller-supplied) is what keeps
+    two submissions from colliding on disk.
+    """
+    if name is None or not name.strip():
+        raise ValidationError("filename is empty.")
+    candidate = name.strip()
+    if any(ch in candidate for ch in _FORBIDDEN_FILENAME_CHARS):
+        raise ValidationError(
+            f"filename must not contain a path separator or a null byte: {name!r}"
+        )
+    if candidate in (".", ".."):
+        raise ValidationError(f"filename must not be '.' or '..': {name!r}")
+    if candidate.startswith("."):
+        raise ValidationError(f"filename must not start with a dot: {name!r}")
+    if Path(candidate).name != candidate:
+        raise ValidationError(f"filename must be a bare name, not a path: {name!r}")
+    if not Path(candidate).stem:
+        raise ValidationError(f"filename has no stem: {name!r}")
+    return candidate
+
+
+def new_submission_id(now: Optional[datetime.datetime] = None) -> str:
+    """`<YYYY-MM-DD>T<HHMMSS>Z-<8 hex>` - sortable, never caller-chosen, so a
+    colliding name can never overwrite a different submission."""
+    moment = now or datetime.datetime.now(datetime.timezone.utc)
+    return f"{moment.strftime('%Y-%m-%dT%H%M%SZ')}-{secrets.token_hex(4)}"
+
+
+def _now_iso() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def _parse_iso(value: Any) -> Optional[datetime.datetime]:
+    if not isinstance(value, str):
+        return None
+    try:
+        parsed = datetime.datetime.fromisoformat(value)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=datetime.timezone.utc)
+    return parsed
+
+
+def _upload_root(root: "Path | str") -> Path:
+    return (Path(root) / "mcp-upload").resolve()
+
+
+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
+    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."""
+    base = _upload_root(root)
+    target = (base / relative).resolve()
+    try:
+        target.relative_to(base)
+    except ValueError:
+        raise ValidationError(f"refusing to write outside mcp-upload/: {relative}")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    tmp = target.with_name(f".{target.name}.{secrets.token_hex(4)}.tmp")
+    tmp.write_bytes(data)
+    os.replace(tmp, target)
+    return target
+
+
+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
+    argument, a human, not the generator above, so it gets the same
+    containment check as a write."""
+    base = _upload_root(root)
+    candidate = (base / submission_id).resolve()
+    try:
+        candidate.relative_to(base)
+    except ValueError:
+        raise ValidationError(f"'{submission_id}' is not a valid submission id.")
+    return candidate
+
+
+def _append_ledger(root: "Path | str", event: dict) -> None:
+    """Append one event to `mcp-upload/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)
+    path = base / "ledger.jsonl"
+    line = json.dumps(event, sort_keys=True) + "\n"
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write(line)
+
+
+def _read_ledger(root: "Path | str") -> list[dict]:
+    path = _upload_root(root) / "ledger.jsonl"
+    if not path.is_file():
+        return []
+    events: list[dict] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            parsed = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(parsed, dict):
+            events.append(parsed)
+    return events
+
+
+def _pending_manifests(root: "Path | str") -> list[dict]:
+    base = _upload_root(root)
+    if not base.is_dir():
+        return []
+    out = []
+    for manifest_path in sorted(base.glob("*/manifest.json")):
+        try:
+            data = json.loads(manifest_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        if isinstance(data, dict):
+            out.append(data)
+    return out
+
+
+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
+    rejection is not blocked by this check."""
+    for manifest in _pending_manifests(root):
+        if manifest.get("sha256") == digest:
+            return manifest.get("id")
+    return None
+
+
+def _check_quota(root: "Path | str", cfg: UploadConfig, 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
+        when = _parse_iso(event.get("time"))
+        if when is None or when < window_start:
+            continue
+        count += 1
+        total_bytes += int(event.get("size") or 0)
+    if count + 1 > cfg.submissions_per_day:
+        raise ValidationError(
+            f"Quota exceeded: '{submitter}' has submitted {count} file(s) in the last 24h "
+            f"(limit {cfg.submissions_per_day}). Nothing was written."
+        )
+    if total_bytes + size > cfg.bytes_per_day:
+        raise ValidationError(
+            f"Quota exceeded: '{submitter}' has submitted {total_bytes} byte(s) in the last 24h "
+            f"(limit {cfg.bytes_per_day} bytes). Nothing was written."
+        )
+
+
+def submit(
+    root: "Path | str",
+    cfg: UploadConfig,
+    *,
+    filename: str,
+    content_b64: str,
+    submitter: Optional[str],
+) -> dict:
+    """Accept one submission into the quarantine, or raise `ValidationError`
+    without writing anything.
+
+    `submitter` must already be the value the identity header carried - this
+    function does not know about HTTP, headers, or which one is configured
+    (`cfg.identity_header` names it only for the refusal message). A `None`
+    or empty `submitter` is refused outright: an unattributable submission is
+    impossible by construction, not merely discouraged.
+    """
+    if not submitter or not submitter.strip():
+        raise ValidationError(
+            f"No identity header ({cfg.identity_header}) on this request - refusing to accept "
+            "an unattributable submission. Nothing was written."
+        )
+    submitter = submitter.strip()
+
+    clean_name = sanitize_filename(filename)
+    ext = Path(clean_name).suffix.lower()
+    if ext not in cfg.allowed_extensions:
+        raise ValidationError(
+            f"'{ext or '(none)'}' is not an allowed extension. Allowed: "
+            f"{', '.join(cfg.allowed_extensions)}"
+        )
+
+    # The base64 length is an upper bound on the decoded size (len*3/4), high
+    # by at most the 0-2 padding characters a valid encoding carries: refuse
+    # before decoding whenever *even the most optimistic reading* still
+    # exceeds the limit, so an oversized submission cannot allocate memory in
+    # its own size just to be measured, without rejecting a legitimate
+    # payload sitting exactly at the limit on padding alone. The post-decode
+    # check below is the exact enforcement; this is only the early exit for
+    # what is unambiguously too large.
+    approx = (len(content_b64 or "") * 3) // 4
+    if approx - 2 > cfg.max_bytes:
+        raise ValidationError(
+            f"Submission is too large (~{approx} bytes, limit {cfg.max_bytes}). "
+            "Nothing was written."
+        )
+    try:
+        data = base64.b64decode(content_b64 or "", validate=True)
+    except binascii.Error as exc:
+        raise ValidationError(f"content_base64 is not valid base64 ({exc}).") from exc
+    if not data:
+        raise ValidationError("Submission is empty. Nothing was written.")
+    if len(data) > cfg.max_bytes:
+        raise ValidationError(
+            f"Submission is too large ({len(data)} bytes, limit {cfg.max_bytes}). "
+            "Nothing was written."
+        )
+
+    digest = hashlib.sha256(data).hexdigest()
+    pending = digest_to_pending_id(root, digest)
+    if pending is not None:
+        raise ValidationError(
+            f"This exact content is already waiting for review as '{pending}'. "
+            "Nothing was written."
+        )
+
+    _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"),
+    )
+    _append_ledger(root, {
+        "event": "submitted",
+        "id": submission_id,
+        "submitter": submitter,
+        "time": manifest["submitted_at"],
+        "size": len(data),
+        "sha256": digest,
+    })
+    return manifest
+
+
+def list_submissions(root: "Path | str") -> list[dict]:
+    """Every manifest currently waiting in the quarantine, oldest id first
+    (the id's own timestamp prefix sorts that way)."""
+    return sorted(_pending_manifests(root), key=lambda m: m.get("id", ""))
+
+
+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/.")
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValidationError(f"Submission '{submission_id}' has a corrupt manifest ({exc}).") from exc
+    if not isinstance(data, dict):
+        raise ValidationError(f"Submission '{submission_id}' has a corrupt manifest.")
+    return data
+
+
+def confirm_token(manifest: dict) -> str:
+    """sha256 over id/filename/size/sha256/submitter, cut to 12 hex chars -
+    same shape as `git_publish.changeset_token`: same submission, same token;
+    anything about it moving (a re-submission under the same id is
+    impossible, but a stale token from an old manifest is not) changes it."""
+    payload = json.dumps(
+        {key: manifest.get(key) for key in ("id", "filename", "size", "sha256", "submitter")},
+        sort_keys=True,
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]
+
+
+def promote(root: "Path | str", submission_id: str) -> Path:
+    """Move a submission's file into `incoming/`, delete its quarantine
+    directory, and append an `accepted` ledger event. The gate (Exit 42, a
+    `--confirm` token) is `commands/upload_cmd.py`'s job, not this function's
+    - by the time this runs, clearance has already happened.
+
+    Every check runs before anything moves: an unknown id, a missing file on
+    disk, or an already-occupied `incoming/<filename>` all refuse with
+    nothing touched. Not atomic across the three effects (move, directory
+    cleanup, ledger append) - the same "one filesystem move, then a write"
+    shape `raw_cmd.raw_accept_command` already has - but every step it does
+    take is ordered so that an interruption leaves file content intact
+    either in the quarantine or in `incoming/`, never neither.
+    """
+    manifest = read_manifest(root, submission_id)
+    src_dir = _pending_dir(root, submission_id)
+    src = src_dir / manifest["filename"]
+    if not src.is_file():
+        raise ValidationError(f"Submission '{submission_id}' is missing its file on disk.")
+
+    dest = Path(root) / "incoming" / manifest["filename"]
+    if dest.exists():
+        raise ValidationError(
+            f"incoming/{manifest['filename']} already exists - rename or clear it first. "
+            f"Nothing was moved for '{submission_id}'."
+        )
+
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    src.rename(dest)
+    shutil.rmtree(src_dir)
+    _append_ledger(root, {
+        "event": "accepted",
+        "id": submission_id,
+        "submitter": manifest.get("submitter"),
+        "time": _now_iso(),
+        "size": manifest.get("size"),
+        "sha256": manifest.get("sha256"),
+    })
+    return dest
+
+
+def reject(root: "Path | str", submission_id: str, reason: str) -> None:
+    """Delete a submission's material, keeping only its ledger trail - the
+    ledger entry is written *before* the delete, so an interruption between
+    the two still leaves the record of why it was rejected."""
+    if not reason or not reason.strip():
+        raise ValidationError("--reason is required and must not be empty.")
+    manifest = read_manifest(root, submission_id)
+    src_dir = _pending_dir(root, submission_id)
+    _append_ledger(root, {
+        "event": "rejected",
+        "id": submission_id,
+        "submitter": manifest.get("submitter"),
+        "time": _now_iso(),
+        "size": manifest.get("size"),
+        "sha256": manifest.get("sha256"),
+        "reason": reason.strip(),
+    })
+    shutil.rmtree(src_dir)