chemenu/tools/chemenu/commands/upload_cmd.py

"""`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), 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`).
"""
from __future__ import annotations

import json as json_module
from typing import Optional

import typer

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="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))


@app.command("list")
@cli_contract.record(cli_contract.CommandRecord(
    path="upload list",
    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,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Read-only",
        budget=cli_contract.Budget.COUNTED,
    ),
    notes=(
        "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.",
    ),
    failures=(),
    examples=(
        "tools/wikitool upload list",
        "tools/wikitool upload list --json",
    ),
    see_also=(
        "`wikitool upload show` - one submission's manifest",
        "`docs/how-to/set-up-the-mcp-server.md` § \"Optional: the submit path\"",
    ),
))
def upload_list_command(
    json_out: bool = typer.Option(False, "--json", help="Print every waiting submission as JSON"),
):
    """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 quarantine/.")
        return
    for manifest in manifests:
        typer.echo(
            f"{manifest['id']}  {manifest['filename']}  {manifest['size']}B  "
            f"via {manifest.get('channel', '?')}  from {manifest['submitter']}"
        )


@app.command("show")
@cli_contract.record(cli_contract.CommandRecord(
    path="upload show",
    summary="Print one submission's manifest in full.",
    synopsis=(cli_contract.Variant(usage="upload show <id> [--json]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.READ,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Read-only",
        budget=cli_contract.Budget.COUNTED,
    ),
    notes=(
        "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.",
    ),
    failures=(cli_contract.Failure(
        cause="Unknown or malformed submission id",
        reaction="Fix the id (see `upload list`) and retry",
    ),),
    examples=(
        "tools/wikitool upload show <id>",
    ),
    see_also=(
        "`wikitool upload accept` - promotes it after review",
        "`wikitool upload reject` - declines it",
    ),
))
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",
                "channel", "handler", "origin"):
        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']}",
        f"  channel:          {manifest.get('channel')}",
        f"  handler:          {manifest.get('handler')}",
        f"  origin:           {manifest.get('origin')}",
        "",
    ]
    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")
@cli_contract.record(cli_contract.CommandRecord(
    path="upload accept",
    summary="**Upload Review Gate:** promote a submission's file from quarantine into `incoming/`.",
    synopsis=(cli_contract.Variant(usage="upload accept <id> [--confirm TOKEN]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.WRITE,
        idempotent=cli_contract.Idempotent.NO,
        atomic="No - one filesystem move, one directory delete, one ledger append; the gate "
        "check runs first, before any of them",
        budget=cli_contract.Budget.COUNTED,
        gates=("upload-review",),
    ),
    notes=(
        "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.",
    ),
    failures=(
        cli_contract.Failure(
            cause="Unknown or malformed submission id, or the submission's file is missing from "
            "`quarantine/<id>/`",
            reaction="Fix the named argument and retry once",
        ),
        cli_contract.Failure(
            cause="`incoming/<filename>` already exists",
            reaction="Not fixed by retrying unchanged - rename or clear it first",
        ),
        cli_contract.Failure(
            cause="Upload Review Gate: `--confirm` is absent or does not match the manifest's "
            "current token",
            reaction=cli_contract.token_gate_reaction("--confirm"),
            code=42,
        ),
    ),
    examples=(
        "tools/wikitool upload accept <id>",
        "tools/wikitool upload accept <id> --confirm <token>    # re-run after exit 42, once "
        "the user approved the manifest",
    ),
    never=(
        "Never pass a `--confirm` token the user has not seen and approved.",
    ),
    see_also=(
        "`wikitool upload show` - the manifest to review",
        "`wikitool raw accept` - the next step for the file in `incoming/`",
        "`instructions/gates.md` - the gate procedure",
    ),
))
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")
@cli_contract.record(cli_contract.CommandRecord(
    path="upload reject",
    summary="Delete a submission's material, keeping only its ledger trail.",
    synopsis=(cli_contract.Variant(usage='upload reject <id> --reason "<why>"'),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.WRITE,
        idempotent=cli_contract.Idempotent.NO,
        atomic="No - one ledger append, then one recursive delete; the ledger write happens "
        "first, so an interruption still leaves the reason on record",
        budget=cli_contract.Budget.COUNTED,
    ),
    notes=(
        "Appends an append-only `rejected` event naming the reason and the sha256 of what was "
        "declined, then deletes the submission's material.",
        "The ledger write happens first, so an interruption still leaves the reason on record.",
        "No gate - rejecting needs no clearance, only accepting does.",
        "Not idempotent: a second call with the same id reports \"unknown id\", which "
        "confirms the first call worked.",
    ),
    failures=(cli_contract.Failure(
        cause="Unknown or malformed submission id, or an empty `--reason`",
        reaction="Fix the argument and retry once. After a first successful call, \"unknown "
        "id\" is confirmation, not a failure",
    ),),
    examples=(
        'tools/wikitool upload reject <id> --reason "duplicate of raw/articles/llm-wiki.md"',
    ),
    see_also=(
        "`wikitool upload show` - the manifest to review",
    ),
))
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}")


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.",
        "`folder` never empties backup or export data: a `path` that holds a `.chemenu-backup.json` "
        "(a machine's folder of the backup, any machine's) or a `.wikitool-export.json` (an "
        "export), is a bare git repository, or lies in one of these, is that channel's error - "
        "nothing is read and nothing removed. A `path` above such folders is fine, since only "
        "its top level is read.",
        "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, a folder in backup or export data, "
            "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. A folder in "
            "backup or export data is the user's to move; never remove a manifest to make it pass",
        ),
    ),
    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)