ingest: bounded parallel ingest (--jobs N), on a context-local wiki redirect

CHANGELOG.md

@@ -8,6 +8,35 @@ All notable changes to this project are documented here. The format is based on
 
 ### Added
 
+- **Bounded parallel ingest — `citadel ingest --jobs N`** (the 2026-07 audit's backlog #11, and
+  with it its finding 1.2.5). Ingest was strictly serial because the per-source staging redirect
+  *assigned* `config.WIKI_DIR` and `os.environ` in place: one process could only ever be inside one
+  redirect. That redirect is now context-local (a `ContextVar` behind `config.wiki_dir()` and
+  friends; child processes get their wiki through an explicit per-spawn env), which is what lets N
+  sources stage at once — the isolation primitive, a per-source staging copy, was already there.
+  `--jobs N` (or `CITADEL_JOBS`) folds in that many sources concurrently. **Every guarantee is
+  unmoved**: one promote per source, all-or-nothing, nothing partial on the live wiki, the manifest
+  still saved per completed source. What is shared is serialized rather than raced — one lock guards
+  the two moments that touch the live wiki (cloning it, promoting onto it), while the minutes-long
+  sessions run fully in parallel; the promote is *base-aware*, pruning only what its own clone had
+  and its staging lacks, so a concurrent source's new page is never deleted as "the agent removed
+  it"; and a promote whose pages have moved since the clone is refused **before** it writes
+  anything, with that source re-run serially at the end of the run, where it sees the winner's page
+  and merges into it (the report lists these under *Re-run serially* — the one place parallelism
+  costs a session a serial run would not have). Report, manifest, and failures bookkeeping stay on
+  the main thread, so those writes are as single-threaded as they ever were, and an interrupt
+  cancels queued sources while still recording work already promoted. The default stays **1** —
+  serial, line for line as before — because the real cost is not safety but **cross-linking**:
+  concurrent sessions cannot see each other's new pages. Best on a large backlog of unrelated
+  sources (see docs/recipes.md), and available on `citadel refresh --jobs N` too — a refresh slice is ordered by last-checked time rather than by topic, so its sources are usually unrelated; `curate` remains serial by design. Chunked sources needed one more
+  rule, found by an adversarial cross-review and reproduced offline before it was fixed: a resume
+  checkpoint's delta is measured against the wiki its source was **cloned from**, never against a
+  live wiki that has since moved on. Measured against live, a page a concurrent source *created*
+  read as a deletion this source had made, and one it *rewrote* read as this source's change
+  carrying stale bytes — and because a checkpoint is durable, that delta outlived the parallel run
+  and could prune a fully-ingested source's page in a later, even strictly serial, run. The base
+  state a replay is guarded against now comes from that same snapshot, so a page another source
+  touched fails the guard instead of passing it.
 - **`citadel serve --http` — the opt-in Streamable HTTP transport** (the 2026-07 audit's backlog
   #12, closing the last open item of its § 3.1 MCP-surface gap). Stdio remains the default and is
   unchanged; `--http` serves the SAME thirteen tools, four prompts, and `wiki://` resources over

CLAUDE.md

@@ -28,9 +28,10 @@ uv run python -m citadel <subcommand>
 Subcommands: `init [DIR]` (scaffold a workspace: `citadel.toml` marker, `.env`, `raw/`, `wiki/`;
 idempotent), `ingest [paths…]` (fold raw/ into the wiki; `--verbose`/`-v` streams the agent
 session, `--log-dir DIR` writes a transcript per source, `--quiet` drops the progress spinner,
+`--jobs N`/`-j` folds N sources in CONCURRENTLY (default 1 = serial; `CITADEL_JOBS`),
 `--full-rescan` distrusts the manifest's stat cache and re-hashes every tracked source,
 `--force <paths>` deliberately re-reads already-ingested sources as a reconcile — it requires
-explicit paths and is refused without them), `refresh [--limit N] [--min-age-days D] [--dry-run]`
+explicit paths and is refused without them), `refresh [--limit N] [--min-age-days D] [--dry-run] [--jobs N]`
 (the THIRD lifecycle: re-verify the least-recently-checked sources — ordered by the manifest's
 `ingested_at` stamp, oldest/stampless first — through forced reconcile sessions on an explicit
 per-run budget of N sources; the sustainable alternative to regenerating the wiki after a model
@@ -229,7 +230,20 @@ it is itself a workspace.
   vanished source's stale provenance before any pending session touches a page that still cites it
   (else that pre-existing bad citation would fail the pending session's validation and roll it
   back). This all-or-nothing + network-share-hardened machinery (`_robust_*`, `robust_mkdir`) is
-  load-bearing — don't simplify it away. **One mutating run per workspace**: ingest and curate take
+  load-bearing — don't simplify it away. **Bounded parallelism** (`--jobs N` / `CITADEL_JOBS`, default 1 = serial): the per-source staging
+  copy IS the isolation primitive, so N sources run at once through the same `_SourceJob` loop —
+  worker threads only plan/stage/run/promote, while every SHARED write (report, manifest, failures)
+  stays on the main thread. One lock (`_LIVE_WIKI_LOCK`) guards the only two moments that touch the
+  live wiki: the clone (taken together with a hash snapshot of what was cloned) and the promote.
+  Under it the promote is **base-aware** — it prunes only what its own clone had and its staging
+  lacks (else it would delete a concurrent source's new pages), and it REFUSES, before writing a
+  byte, if a path it would touch has moved since the clone; that source is then re-run SERIALLY at
+  the end of its group, where it merges into the winner's page (reported as `raced`). What
+  parallelism costs is cross-linking, not safety — concurrent sessions cannot see each other's new
+  pages — which is why the default is 1. The redirect this rides on is context-local
+  (`config.wiki_redirect` → a ContextVar behind `config.wiki_dir()`; children get their wiki via
+  `config.child_env()`), never a `config.WIKI_DIR`/`os.environ` assignment.
+  **One mutating run per workspace**: ingest and curate take
   an exclusive run lock (`runlock.py`, a dotfile sibling of the wiki; stale locks reclaimed via
   dead-pid/mtime, refreshed per source) so a second concurrent run fails loud instead of silently
   destroying the first one's staging/promotes; manifest + failures saves are atomic
@@ -408,8 +422,12 @@ save-the-transcript-as-a-file lane for whole conversations). `rawsource.py` back
 
 ## Conventions specific to this codebase
 
-- **`config.*` is read at call time** (`from . import config` then `config.WIKI_DIR`), never imported
+- **`config.*` is read at call time** (`from . import config` then `config.RAW_DIR`), never imported
   by value — so tests can monkeypatch the whole filesystem layout. Honor this when adding code.
+  The WIKI path is read through the ACCESSORS (`config.wiki_dir()`, `index_path()`,
+  `sources_index_path()`, `log_path()`, `manifest_path()`, `failures_path()`), never as
+  `config.WIKI_DIR`: ingest redirects it per source through a ContextVar, so only the accessors see
+  the staging copy. The module attributes stay the process-wide base tests monkeypatch.
 - **Tests redirect everything to `tmp_path`** by monkeypatching `config.*` (including
   `WORKSPACE_ROOT`, which the agent's `cwd` reads) and replace `llm.run_ingest_session` with a fake
   that writes files into the temp wiki. No test spawns a real LLM CLI. Follow that pattern; keep
@@ -437,7 +455,8 @@ save-the-transcript-as-a-file lane for whole conversations). `rawsource.py` back
   `READ_ONLY`), `CITADEL_LLM_VERBOSE`, `CITADEL_LLM_LOG_DIR`,
   `CITADEL_REPO_SUPPORT`, `CITADEL_IMAGE_SUPPORT` (read images visually), `CITADEL_AUDIO_SUPPORT`
   (opt-in whisper transcript ingest for audio/video, with `CITADEL_WHISPER_CLI`/
-  `CITADEL_WHISPER_MODEL`/`CITADEL_WHISPER_TIMEOUT` tuning the seam), `CITADEL_MAX_SOURCE_CHARS`
+  `CITADEL_WHISPER_MODEL`/`CITADEL_WHISPER_TIMEOUT` tuning the seam), `CITADEL_JOBS` (how many sources ingest folds in
+  concurrently; 1 = serial), `CITADEL_MAX_SOURCE_CHARS`
   (large-source chunking threshold), `CITADEL_RESUME` (resume checkpoints for those chunked
   sources: continue at the segment an interrupted run died on instead of re-paying for the earlier
   ones; default on), `CITADEL_DEDUP_BY_BASENAME` (skip same-basename document

citadel/catalogs.py

@@ -192,7 +192,7 @@ def rebuild_indexes(pages: list[Page] | None = None) -> None:
     # separate O(sources × pages) scans the two consumers used to run independently.
     refs_by_key = citing_pages_map(manifest_dict, pages)
     sources_body = _render_sources_catalog(manifest_dict, pages, failures_dict, refs_by_key)
-    sources_path = config.SOURCES_INDEX_PATH
+    sources_path = config.sources_index_path()
     if sources_body is not None:
         config.robust_mkdir(sources_path.parent)
         sources_path.write_text(sources_body, encoding="utf-8")
@@ -204,7 +204,7 @@ def rebuild_indexes(pages: list[Page] | None = None) -> None:
     # sources catalog it is written when there is at least one point and removed when there are none.
     open_points = collect_open_points(pages)
     open_points_body = _render_open_points_catalog(open_points, title_by_path)
-    open_points_path = config.WIKI_DIR / OPEN_POINTS_INDEX_REL
+    open_points_path = config.wiki_dir() / OPEN_POINTS_INDEX_REL
     if open_points_body is not None:
         config.robust_mkdir(open_points_path.parent)
         open_points_path.write_text(open_points_body, encoding="utf-8")
@@ -303,8 +303,8 @@ def rebuild_indexes(pages: list[Page] | None = None) -> None:
         lines.append("")
 
     body = "\n".join(lines).rstrip("\n") + "\n"
-    config.robust_mkdir(config.WIKI_DIR)
-    (config.WIKI_DIR / "index.md").write_text(body, encoding="utf-8")
+    config.robust_mkdir(config.wiki_dir())
+    (config.wiki_dir() / "index.md").write_text(body, encoding="utf-8")
 
     # ----- per-directory index.md (also frontmatter-free) -----
     for folder in sorted(folders):
@@ -315,6 +315,6 @@ def rebuild_indexes(pages: list[Page] | None = None) -> None:
             flines.append(f"- [{page.title}]({rel_in_folder}) — {page.description}")
         flines.append("")
         fbody = "\n".join(flines).rstrip("\n") + "\n"
-        folder_dir = config.WIKI_DIR / folder
+        folder_dir = config.wiki_dir() / folder
         config.robust_mkdir(folder_dir)
         (folder_dir / "index.md").write_text(fbody, encoding="utf-8")

citadel/cli.py

@@ -116,6 +116,18 @@ def build_parser() -> argparse.ArgumentParser:
         "without them, so a whole-corpus re-read (one agent session per source) can never "
         "happen by accident.",
     )
+    p_ingest.add_argument(
+        "--jobs",
+        "-j",
+        type=int,
+        default=None,
+        metavar="N",
+        help="Fold up to N sources CONCURRENTLY (default 1 = serial, or CITADEL_JOBS). Each "
+        "source keeps its own staging copy and its own all-or-nothing promote; promotes are "
+        "serialized, and a source that raced another over the same page is re-run serially. "
+        "Faster on a large backlog of unrelated sources; the trade-off is cross-linking, since "
+        "concurrent sessions cannot see each other's new pages.",
+    )
     p_ingest.set_defaults(func=cmd_ingest)
 
     p_curate = sub.add_parser(
@@ -188,6 +200,17 @@ def build_parser() -> argparse.ArgumentParser:
         help="Write a transcript file per source to DIR (see `citadel ingest --log-dir`). "
         "Overrides CITADEL_LLM_LOG_DIR.",
     )
+    p_refresh.add_argument(
+        "--jobs",
+        "-j",
+        type=int,
+        default=None,
+        metavar="N",
+        help="Re-verify up to N sources CONCURRENTLY (default 1 = serial, or CITADEL_JOBS) — the "
+        "same knob, and the same trade-off, as `citadel ingest --jobs`. A refresh slice is ordered "
+        "by last-checked time rather than by topic, so its sources are usually unrelated and "
+        "parallelism costs little cross-linking here.",
+    )
     p_refresh.set_defaults(func=cmd_refresh)
 
     p_status = sub.add_parser(
@@ -388,9 +411,16 @@ def cmd_ingest(args: argparse.Namespace) -> int:
 
     ``--force`` requires explicit paths: a forced re-read runs one
     agent session per source, so forcing the ENTIRE corpus must never happen by accident — the
-    flag alone is refused with exit 2, before ``ingest.ingest`` is ever called."""
+    flag alone is refused with exit 2, before ``ingest.ingest`` is ever called.
+
+    ``--jobs N`` is a usage error below 1 (exit 2, like ``--force`` without paths) rather than an
+    exception out of the API layer; omitted, the run takes ``CITADEL_JOBS`` (default 1, serial)."""
     from . import config, ingest
 
+    if args.jobs is not None and args.jobs < 1:
+        print(f"error: --jobs must be at least 1 (got {args.jobs}); 1 means the serial default.", file=sys.stderr)
+        return 2
+
     if args.force and not args.paths:
         print(
             "error: --force requires explicit paths (a forced re-read runs one agent session per "
@@ -414,7 +444,9 @@ def cmd_ingest(args: argparse.Namespace) -> int:
         # CITADEL_LLM_VERBOSE — not just the --verbose flag — also drops the spinner that would
         # otherwise clobber the streamed transcript.
         progress = ConsoleProgress(spinner=not config.LLM_VERBOSE)
-    report = ingest.ingest(args.paths or None, progress=progress, full_rescan=args.full_rescan, force=args.force)
+    report = ingest.ingest(
+        args.paths or None, progress=progress, full_rescan=args.full_rescan, force=args.force, jobs=args.jobs
+    )
     print(report.render())
     # Non-zero on a per-source error OR a structural problem left behind (a broken
     # cross-link the agent introduced) — so ingest gates the wiki's integrity in CI.
@@ -450,6 +482,9 @@ def cmd_refresh(args: argparse.Namespace) -> int:
             file=sys.stderr,
         )
         return 2
+    if args.jobs is not None and args.jobs < 1:
+        print(f"error: --jobs must be at least 1 (got {args.jobs}); 1 means the serial default.", file=sys.stderr)
+        return 2
     if args.verbose:
         config.LLM_VERBOSE = True
     if args.log_dir is not None:
@@ -459,7 +494,9 @@ def cmd_refresh(args: argparse.Namespace) -> int:
         from .progress import ConsoleProgress
 
         progress = ConsoleProgress(spinner=not config.LLM_VERBOSE)
-    report = refresh.refresh(limit=args.limit, min_age_days=args.min_age_days, dry_run=args.dry_run, progress=progress)
+    report = refresh.refresh(
+        limit=args.limit, min_age_days=args.min_age_days, dry_run=args.dry_run, progress=progress, jobs=args.jobs
+    )
     print(report.render(), end="")
     ing = report.ingest_report
     return 1 if ing is not None and (ing.errors or ing.broken_links) else 0
@@ -719,7 +756,7 @@ def cmd_check(args: argparse.Namespace) -> int:
     issues = validate.validate_all(pages)
     missing: list[str] = []
     if args.paths:
-        wiki_root = config.WIKI_DIR.resolve()
+        wiki_root = config.wiki_dir().resolve()
         wanted: set[str] = set()
         for arg in args.paths:
             rel = arg.replace(os.sep, "/")
@@ -735,7 +772,7 @@ def cmd_check(args: argparse.Namespace) -> int:
         known = {p.rel_path for p in pages}
         for rel in sorted(wanted - known):
             try:
-                on_disk = okf.safe_join(config.WIKI_DIR, rel).is_file()
+                on_disk = okf.safe_join(config.wiki_dir(), rel).is_file()
             except okf.OKFError:
                 on_disk = False
             if on_disk: