raw accept: Stem-Eindeutigkeit im Typverzeichnis erzwingen, --replaces als einziger Weg daran vorbei (schliesst #64)

Files changed:
- CHANGES.md
- VERSION
- instructions/wiki-ingest/SKILL.md
- raw/CONTRACT.md
- tools/CONTRACT.md
- tools/chemenu/commands/raw_cmd.py
- tools/chemenu/tests/test_raw_cmd.py

tools/chemenu/commands/raw_cmd.py

@@ -20,6 +20,20 @@ that is this command's job:
 Multi-owner raw files (`provenance.duplicate_raw_file_owners`) are refused
 rather than silently moved: relocating a file another page also claims would
 break that page's `raw_files:` without it ever being consulted.
+
+**Stem uniqueness at `raw/<type>/` level** (Gitea #64) closes the gap this
+leaves open: without it, a second, unrelated source whose primary file happens
+not to collide on the exact filename slips silently into an existing bundle,
+because the per-file `dst.exists()` check above never looks at the bundle
+directory itself. The set of names occupied at `raw/<type>/` level - file
+stems and bundle directory names alike - must stay unique; `_occupied_stems()`
+and the check built on it enforce that, while still allowing a call to grow a
+bundle it already owns (via `--page`, or by continuing an existing bundle).
+
+**`--replaces <raw-path>`** is the only sanctioned way past that rule: whether
+a new file is a later edition of an existing source or a second, separate one
+is a human's decision, never the tool's or an agent's, so the command refuses
+and names both routes rather than choosing one (Gitea #64 decision 2).
 """
 from __future__ import annotations
 
@@ -33,7 +47,7 @@ from chemenu.commands._util import fail, rel_path, success
 from chemenu.commands.dist_cmd import RAW_SUBDIRS
 from chemenu.frontmatter_io import write_page
 from chemenu.kb_scan import load_kb_pages
-from chemenu.provenance import source_pages_by_raw_file, source_raw_files
+from chemenu.provenance import citing_pages, source_pages_by_raw_file, source_raw_files
 
 app = typer.Typer(help="Promote raw material out of incoming/ into raw/.")
 
@@ -70,6 +84,106 @@ def _classify(path: Path, incoming: Path) -> str:
     return sub
 
 
+def _occupied_stems(raw_sub_dir: Path) -> dict[str, Path]:
+    """Every name occupied at `raw/<type>/` level: file stems and bundle
+    directory names alike, one level below `raw_sub_dir` only."""
+    occupied: dict[str, Path] = {}
+    if raw_sub_dir.is_dir():
+        for entry in raw_sub_dir.iterdir():
+            occupied[entry.stem if entry.is_file() else entry.name] = entry
+    return occupied
+
+
+def _stem_collision_message(sub: str, claimed_name: str, holder: Path) -> str:
+    example_target = holder
+    if holder.is_dir():
+        children = sorted(holder.iterdir())
+        example_target = children[0] if children else holder
+    return (
+        f'{rel_path(holder)} already claims the stem "{claimed_name}" in raw/{sub}/.\n'
+        "  These are two different intents and only you can tell them apart:\n"
+        f"    Same source, new edition   -> tools/wikitool raw accept "
+        f"--replaces {rel_path(example_target)} <incoming file>\n"
+        "    A second, separate source  -> rename it in incoming/ (add a distinguishing "
+        "suffix) and accept it normally\n"
+        "  raw accept does not guess which one this is."
+    )
+
+
+def _replace(files: list[Path], replaces: Path, page: Optional[str], dry_run: bool) -> None:
+    """`raw accept --replaces <target> <incoming file>` - overwrite an
+    existing raw/ file wholesale with a later edition, in place, with
+    `raw_files:` on every source page left untouched (Gitea #64 decision 2).
+
+    Every check below runs before any write, so a failure leaves both the
+    target and the incoming file exactly as they were - the invariant the
+    issue names for this destructive step.
+    """
+    if page is not None:
+        fail("--replaces and --page cannot be combined - a replacement never changes raw_files:.")
+    if len(files) != 1:
+        fail("--replaces takes exactly one incoming file - a replacement is one file for one file.")
+
+    incoming_path = _resolve(files[0])
+    if not incoming_path.is_file():
+        fail(f"{rel_path(incoming_path)} does not exist or is not a file.")
+    incoming_sub = _classify(incoming_path, _incoming_dir())
+
+    target = _resolve(replaces)
+    try:
+        target_rel = target.relative_to(config.RAW_DIR)
+    except ValueError:
+        fail(f"--replaces target {rel_path(target)} does not lie under raw/.")
+    if not target.is_file():
+        fail(f"--replaces target {rel_path(target)} does not exist or is not a file.")
+    target_sub = target_rel.parts[0]
+    if target_sub != incoming_sub:
+        fail(
+            f"--replaces target is under raw/{target_sub}/, but {rel_path(incoming_path)} is "
+            f"under incoming/{incoming_sub}/ - a replacement stays within one type directory."
+        )
+    if target.name != incoming_path.name:
+        fail(
+            f"--replaces target {rel_path(target)} has a different filename than "
+            f"{rel_path(incoming_path)} - a rename is not part of a replacement (see `raw rename`, #16)."
+        )
+
+    pages = load_kb_pages(config.KB_DIR)
+    by_raw = source_pages_by_raw_file(pages)
+    owners = sorted(set(by_raw.get(rel_path(target), [])))
+    if len(owners) > 1:
+        fail(
+            f"Cannot replace {rel_path(target)}: more than one source page claims it "
+            f"({', '.join(owners)}). Resolve the multiple ownership first (see `wikitool sources coverage`)."
+        )
+
+    if dry_run:
+        typer.echo(f"[dry-run] would replace {rel_path(target)} with {rel_path(incoming_path)}. No files written.")
+        return
+
+    target.unlink()
+    incoming_path.rename(target)
+
+    if not owners:
+        success(
+            f"Replaced {rel_path(target)} (previous version stays in git history). "
+            "No source page covers this file - see `wikitool sources coverage`."
+        )
+        return
+
+    source_title = owners[0]
+    citers = citing_pages(pages, source_title)
+    typer.echo(f"Replaced {rel_path(target)} (previous version stays in git history).")
+    typer.echo("These pages were compiled against the previous version and may now be stale:")
+    typer.echo(f"  [[{source_title}]]")
+    if citers:
+        for citer in citers:
+            typer.echo(f"    cited by: [[{citer}]]")
+    else:
+        typer.echo("    cited by: (nothing yet)")
+    success("Review them in this same run: the replacement and their update belong in one commit.")
+
+
 @app.command("accept")
 def raw_accept_command(
     files: list[Path] = typer.Argument(
@@ -82,6 +196,12 @@ def raw_accept_command(
         help="Extend this existing source page's raw_files: with the promoted file(s), "
         "folding in its already-promoted file if this raises it past one",
     ),
+    replaces: Optional[Path] = typer.Option(
+        None,
+        "--replaces",
+        help="Overwrite this existing raw/ file in place with the single incoming/ file passed "
+        "alongside it - the only sanctioned way past the stem-uniqueness rule below",
+    ),
     dry_run: bool = typer.Option(False, "--dry-run", help="List what would move without writing"),
 ):
     """Promote file(s) from incoming/ into raw/, computing the destination
@@ -91,6 +211,11 @@ def raw_accept_command(
         fail("Pass at least one file to promote.")
 
     incoming = _incoming_dir()
+
+    if replaces is not None:
+        _replace(files, replaces, page, dry_run)
+        return
+
     resolved = [_resolve(f) for f in files]
 
     for path in resolved:
@@ -168,6 +293,22 @@ def raw_accept_command(
         if dst.exists():
             fail(f"Cannot promote: {rel_path(dst)} already exists.")
 
+    # Stem uniqueness at raw/<type>/ level (Gitea #64): the name this call is
+    # about to claim there - the bundle's name, or the lone file's stem when no
+    # bundle forms - must not already belong to something this call does not
+    # itself own. "Owns" means: one of the page's already-registered raw files
+    # (the pitfall from the module docstring - a single file growing into a
+    # bundle of its own name momentarily still occupies that name), or, once a
+    # bundle already has >=2 registered files, the bundle directory itself.
+    claimed_name = bundle_dir.name if bundle_dir is not None else resolved[0].stem
+    occupied = _occupied_stems(raw_sub_dir)
+    owned = set(existing_raw_paths)
+    if len(existing_raw_paths) >= 2:
+        owned.add(bundle_dir)
+    holder = occupied.get(claimed_name)
+    if holder is not None and holder not in owned:
+        fail(_stem_collision_message(sub, claimed_name, holder))
+
     moving_existing = [src for src, _dst in moves if src in existing_raw_paths]
     if moving_existing:
         by_raw = source_pages_by_raw_file(pages)

tools/chemenu/tests/test_raw_cmd.py

@@ -20,8 +20,8 @@ def tree(kb_dir):
     return root
 
 
-def _accept(*files, page=None, dry_run=False):
-    return raw_accept_command(files=list(files), page=page, dry_run=dry_run)
+def _accept(*files, page=None, replaces=None, dry_run=False):
+    return raw_accept_command(files=list(files), page=page, replaces=replaces, dry_run=dry_run)
 
 
 def _write_source(kb_dir, title, raw_files):
@@ -254,3 +254,228 @@ def test_type_directory_mismatch_with_existing_raw_files_is_rejected(tree):
     with pytest.raises(typer.Exit):
         _accept(new_file, page="Source - Handbuch")
     assert new_file.exists()
+
+
+# --- Stem uniqueness at raw/<type>/ level (Gitea #64) ---
+
+
+def test_second_source_does_not_silently_join_an_existing_bundle(tree):
+    """The exact repro from #64: a bundle exists for one source, and a second,
+    unrelated source's files (no --page) happen not to collide on filename -
+    they must not slip into the existing bundle."""
+    bundle = tree / "raw/documents/handbuch"
+    bundle.mkdir()
+    (bundle / "handbuch.pdf").write_bytes(b"pdf")
+    (bundle / "handbuch.md").write_text("md", encoding="utf-8")
+
+    txt = tree / "incoming/documents/handbuch.txt"
+    anhang = tree / "incoming/documents/anhang.md"
+    txt.write_text("txt", encoding="utf-8")
+    anhang.write_text("anhang", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(txt, anhang)
+
+    assert txt.exists() and anhang.exists()
+    assert sorted(p.name for p in bundle.iterdir()) == ["handbuch.md", "handbuch.pdf"]
+
+
+def test_flat_promote_rejected_when_stem_matches_an_existing_bundle(tree):
+    (tree / "raw/documents/handbuch").mkdir()
+    (tree / "raw/documents/handbuch/handbuch.pdf").write_bytes(b"pdf")
+
+    txt = tree / "incoming/documents/handbuch.txt"
+    txt.write_text("txt", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(txt)
+
+    assert txt.exists()
+    assert (tree / "raw/documents/handbuch/handbuch.pdf").exists()
+
+
+def test_flat_promote_rejected_when_stem_matches_an_existing_flat_file(tree):
+    """Same stem, different extension: not caught by the old per-path
+    dst.exists() check, which is the whole gap #64 closes."""
+    (tree / "raw/documents/handbuch.pdf").write_bytes(b"pdf")
+
+    md = tree / "incoming/documents/handbuch.md"
+    md.write_text("md", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(md)
+
+    assert md.exists()
+    assert (tree / "raw/documents/handbuch.pdf").read_bytes() == b"pdf"
+    assert not (tree / "raw/documents/handbuch.md").exists()
+
+
+def test_stem_collision_message_names_both_routes_without_recommending_one(tree, capsys):
+    (tree / "raw/documents/handbuch.pdf").write_bytes(b"pdf")
+    md = tree / "incoming/documents/handbuch.md"
+    md.write_text("md", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(md)
+
+    err = capsys.readouterr().out
+    assert "--replaces" in err
+    assert "rename" in err
+    assert "does not guess" in err
+
+
+# --- --replaces (Gitea #64 decision 2) ---
+
+
+def test_replaces_swaps_bytes_and_leaves_raw_files_untouched(tree):
+    _write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old edition", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new edition", encoding="utf-8")
+
+    _accept(new, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "new edition"
+    assert not new.exists()
+    frontmatter, _ = read_page(tree / "kb/sources/Source - Handbuch.md")
+    assert frontmatter["raw_files"] == ["raw/documents/handbuch.md"]
+
+
+def test_replaces_reports_source_and_both_citing_pages(tree, capsys):
+    """AC: one source page, two citing pages - all three named in the output."""
+    _write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
+    write_page(
+        tree / "kb/concepts/Handbuch-Konzept.md",
+        {
+            "type": "types/concept.md", "concept_type": "protocol", "tags": [],
+            "created": "2026-09-01", "modified": "2026-09-01", "related": [],
+            "sources": ["Source - Handbuch"], "confidence": 0.7,
+        },
+        "\n# Handbuch-Konzept\n\n## Definition\n\nx.\n",
+    )
+    write_page(
+        tree / "kb/entities/tools/handbuch-tool.md",
+        {
+            "type": "types/entity.md", "entity_type": "tool", "tags": [],
+            "created": "2026-09-01", "modified": "2026-09-01", "related": [],
+            "sources": ["Source - Handbuch"], "confidence": 0.7,
+        },
+        "\n# handbuch-tool\n\n## Description\n\nx.\n",
+    )
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    _accept(new, replaces=target)
+
+    out = capsys.readouterr().out
+    assert "Source - Handbuch" in out
+    assert "Handbuch-Konzept" in out
+    assert "handbuch-tool" in out
+
+
+def test_replaces_succeeds_with_no_owner_and_reports_it(tree, capsys):
+    target = tree / "raw/documents/orphan.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/orphan.md"
+    new.write_text("new", encoding="utf-8")
+
+    _accept(new, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "new"
+    out = capsys.readouterr().out
+    assert "source page covers this file" in out
+
+
+def test_replaces_dry_run_writes_nothing(tree):
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    _accept(new, replaces=target, dry_run=True)
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert new.exists()
+
+
+def test_replaces_rejects_more_than_one_incoming_file(tree):
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    a = tree / "incoming/documents/handbuch.md"
+    b = tree / "incoming/documents/extra.md"
+    a.write_text("a", encoding="utf-8")
+    b.write_text("b", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(a, b, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert a.exists() and b.exists()
+
+
+def test_replaces_rejects_filename_mismatch(tree):
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch-v2.md"
+    new.write_text("new", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(new, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert new.exists()
+
+
+def test_replaces_rejects_type_directory_mismatch(tree):
+    target = tree / "raw/notes/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(new, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert new.exists()
+
+
+def test_replaces_rejects_nonexistent_target(tree):
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(new, replaces=tree / "raw/documents/handbuch.md")
+
+    assert new.exists()
+
+
+def test_replaces_rejects_multi_owner_target(tree):
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    _write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
+    _write_source(tree / "kb", "Source - Also Handbuch", ["raw/documents/handbuch.md"])
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(new, replaces=target)
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert new.exists()
+
+
+def test_replaces_rejects_combination_with_page(tree):
+    _write_source(tree / "kb", "Source - Handbuch", ["raw/documents/handbuch.md"])
+    target = tree / "raw/documents/handbuch.md"
+    target.write_text("old", encoding="utf-8")
+    new = tree / "incoming/documents/handbuch.md"
+    new.write_text("new", encoding="utf-8")
+
+    with pytest.raises(typer.Exit):
+        _accept(new, replaces=target, page="Source - Handbuch")
+
+    assert target.read_text(encoding="utf-8") == "old"
+    assert new.exists()