diff --git a/.kiro/specs/contributor-scan/tasks.md b/.kiro/specs/contributor-scan/tasks.md
index 128b4f4..25c0e82 100644
--- a/.kiro/specs/contributor-scan/tasks.md
+++ b/.kiro/specs/contributor-scan/tasks.md
@@ -45,7 +45,7 @@
   - _Requirements: 3.1, 3.2, 3.3, 3.4, 3.5, 8.3_
   - _Depends: 1.1_
 
-- [ ] 1.3 The MAP stage
+- [x] 1.3 The MAP stage
   - Wire `scan_repository` into `cli.py` inside `Stage.MAP`, mirroring `_check_obligations`: takes
     `entry_points` and `targets`, returns findings merged **additively** beside the pattern, overlay and
     obligation findings. A `[model]` config block (enabled, budget, max regions), zero-dep, conservative
@@ -58,6 +58,25 @@
   - _Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 8.4_
   - _Depends: 1.2_
 
+- [ ] 1.4 Batch the queries — one Joern invocation per scan, not one per region and family
+  - **Why this is in group 1 rather than group 2.** Wiring 1.3 exposed that the cost model in the design was
+    half right. One CPG per scan was the right call, but every *query* is its own `joern --script` JVM launch
+    of roughly 30 seconds, and the driver issues one per region per family. A single ten-line Python file
+    admits six families and therefore takes six launches and over three minutes. At 100 regions that is ~5
+    hours; at 1000, ~50. Task 2.2's go/no-go would not have been a measurement, only arithmetic.
+  - Extend `cpg/backend.py` with a batched call: one invocation carrying many (region, family, spec) requests
+    and returning rows keyed back to each. The `.sc` scripts already take parameter lists; they take a list of
+    requests instead. `model/scan.py` collects the work first, issues one call, then arbitrates from the rows.
+  - **Keep the seam.** The subprocess boundary stays the only Joern touch-point (predecessor Req 4.5). Server
+    mode would keep a JVM warm but adds a process model and weakens that promise; batching needs neither.
+  - **The arbiters do not change.** `taint`, `dominance`, `config_value` and `pipeline` take rows and are
+    already measured; if this task needs to edit one, that is a signal the batching is leaking into them.
+  - Observable: a scan over N regions issues **one** `joern --script` invocation, not N×families; the same
+    fixture produces the same findings and rungs as the unbatched path, asserted against a recorded baseline;
+    the ten-line scan drops from minutes to seconds.
+  - _Requirements: 3.5, 4.2, 4.3_
+  - _Depends: 1.3_
+
 ## Group 2 — Scale it (the phase that can reshape the rest)
 
 - [ ] 2.1 A pinned known-vulnerable checkout
@@ -151,6 +170,9 @@
 
 - Group 1 must land and pass review before group 2: measuring an unwired pipeline is what produced a
   session's worth of harness-only numbers.
+- Task 1.4 was added after 1.3 was wired, when a ten-line scan took three minutes. It belongs in group 1
+  because group 1's promise is a working scan, and a scan that cannot finish on a real repository is not one.
+  Finding it here rather than in 2.2 is the argument for wiring before measuring.
 - **Task 2.2 is a go/no-go.** If one CPG per repository is infeasible, stop and redesign the region/budget
   model rather than building groups 3–6 on an assumption.
 - Group 5 is deliberately last. The corpus is not the binding constraint today — reachability is — and 198 of
diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py
index e3113a4..f3e030e 100644
--- a/src/openultrasast/cli.py
+++ b/src/openultrasast/cli.py
@@ -31,7 +31,7 @@ from .calibration import (
 )
 from .complexity import map as complexity_map
 from .complexity.ledger import persist_verdicts
-from .config import ObligationsConfig, load_config, load_dotenv
+from .config import ModelLayerConfig, ObligationsConfig, ResolvedConfig, load_config, load_dotenv
 from .findings import StaticFinding, quick_scan_findings, write_findings
 from .fusion import FusionDecision, fuse_findings_dispatch
 from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR
@@ -390,6 +390,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     overlay_records: list[OverlayRecord] = []
     wrote_overlay = False
     obligations_payload: dict[str, object] | None = None
+    model_payload: dict[str, object] | None = None
     obligations_cited: dict[str, dict[str, object]] = {}
     if Stage.MAP in plan.requested:
         built_map = runtime.run_stage(
@@ -436,6 +437,19 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             if obligation_findings:
                 findings = rank_obligations(findings + obligation_findings)
                 write_findings(findings, findings_path)
+        # Fifth MAP proposer: the model layer (contributor-scan Req 1). Additive -- it adds findings beside
+        # the pattern, overlay and obligation ones and removes none of them.
+        model_findings, model_payload = _run_model_layer(
+            root=run.target,
+            targets=targets,
+            entries=entry_points,
+            settings=config.model,
+            runtime=runtime,
+            config=config,
+        )
+        if model_findings:
+            findings = findings + model_findings
+            write_findings(findings, findings_path)
         plan = record_completed(plan, Stage.MAP)
         if not hunter_model:
             runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable"))
@@ -554,6 +568,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             worth_fixing=worth_fixing_payload,
             provenance=provenance.to_dict(),
             obligations=obligations_payload,
+            model=model_payload,
         ),
     )
     runtime.finish(status="succeeded")
@@ -788,6 +803,118 @@ def _pairs(
     return 0
 
 
+def _model_skipped(reason: str) -> dict[str, object]:
+    """A skipped stage reports the same SHAPE as a run one, plus why it skipped.
+
+    A consumer reading this should not have to branch on whether the stage ran; a payload that is sometimes
+    a full record and sometimes a single key is how a report ends up silently omitting a section.
+    """
+    return {**_model_payload(None), "skipped": reason}
+
+
+def _model_payload(result: object, *, unjudged_paths: tuple[str, ...] = ()) -> dict[str, object]:
+    """What the scan records about the model layer.
+
+    `unjudged_sample` names the regions the budget did not reach, lowest-ranked first. Counting them is the
+    honesty requirement (Req 3.3); NAMING them is what makes an exclusion list for a large repository
+    something a maintainer can write from evidence rather than guesswork.
+    """
+    return {
+        "by_rung": dict(getattr(result, "by_rung", {}) or {}),
+        "regions_scanned": getattr(result, "regions_scanned", 0),
+        "regions_unjudged": getattr(result, "regions_unjudged", 0),
+        "unjudged_sample": list(unjudged_paths),
+        "model_calls": getattr(result, "model_calls", 0),
+        "cost_usd": getattr(result, "cost_usd", 0.0),
+        "seconds": getattr(result, "seconds", 0.0),
+    }
+
+
+def _run_model_layer(
+    *,
+    root: Path,
+    targets: list[FileTarget],
+    entries: Sequence[object],
+    settings: ModelLayerConfig,
+    runtime: HarnessRuntime,
+    config: ResolvedConfig,
+) -> tuple[list[StaticFinding], dict[str, object]]:
+    """Fifth MAP proposer (contributor-scan Req 1): the model layer, arbitrating regions of the repository.
+
+    Additive by construction. Without a CPG engine this records a reason and returns nothing, and the scan is
+    byte-for-byte what it is today -- the model layer must never make the tool worse for someone who does not
+    have Joern.
+    """
+    if not settings.enabled:
+        return [], _model_skipped("disabled")
+
+    from .cpg.backend import resolve_cpg_backend
+    from .model.endpoint import resolve_chat_endpoint, resolve_models
+    from .model.regions import regions_for
+    from .model.scan import ScanBudget, scan_repository
+
+    backend = resolve_cpg_backend()
+    if not backend.available():
+        runtime.state["degradations"].append({"stage": "model", "reason": "cpg_unavailable"})
+        return [], _model_skipped("cpg_unavailable")
+
+    regions = regions_for(entries, targets)
+    if not regions:
+        return [], _model_skipped("no_regions")
+
+    resolved = resolve_chat_endpoint(config)
+    client = resolved[0] if resolved else None
+    if client is None:
+        # Not a failure: entailment still works and only the suspicion band goes unasked (Req 11.3).
+        runtime.state["degradations"].append({"stage": "model", "reason": "learning_endpoint_unavailable"})
+
+    result = runtime.run_stage(
+        "model",
+        lambda: scan_repository(
+            root,
+            regions,
+            backend=backend,
+            client=client,
+            model=resolve_models(config)[0],
+            budget=ScanBudget(max_model_calls=settings.max_model_calls, max_regions=settings.max_regions),
+        ),
+    )
+    for degradation in result.degradations:
+        runtime.state["degradations"].append(dict(degradation))
+
+    # The regions the budget did not reach, weakest first -- the evidence an exclusion list is built from.
+    unjudged = tuple(region.path for region in regions[result.regions_scanned :][:10])
+    findings = [_finding_from_model(item) for item in result.findings]
+    return findings, _model_payload(result, unjudged_paths=unjudged)
+
+
+def _finding_from_model(item: object) -> StaticFinding:
+    """A model finding as a StaticFinding, carrying its rung and witness (Req 5.1)."""
+    site = str(getattr(item, "site", ""))
+    path, _, rest = site.partition(":")
+    line = rest.split(":")[0] if rest else ""
+    rung = getattr(item, "rung", None)
+    witness = str(getattr(item, "witness", "") or getattr(item, "contradiction", ""))
+    family = str(getattr(item, "family", ""))
+    return StaticFinding(
+        finding_id=f"model:{family}:{site}",
+        path=path or site,
+        title=f"{family} ({getattr(rung, 'value', rung)})",
+        severity="medium",
+        confidence="medium",
+        evidence_level="suspicion",
+        rationale=witness or f"the model layer reported a {family} candidate",
+        line=int(line) if line.isdigit() else None,
+        function_name=None,
+        reachability_status="unknown",
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=[f"family:{family}", f"rung:{getattr(rung, 'value', rung)}"],
+        ranking_priority=0.0,
+        rung=str(getattr(rung, "value", rung) or "suspicion"),
+    )
+
+
 def _check_obligations(
     *,
     root: Path,
diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py
index 69d2253..805970d 100644
--- a/src/openultrasast/config.py
+++ b/src/openultrasast/config.py
@@ -143,6 +143,18 @@ class VariantsConfig:
     max_mechanisms: int = 500
 
 
+@dataclass(frozen=True)
+class ModelLayerConfig:
+    # contributor-scan: the model layer in MAP. Named for the LAYER, not the model: `ModelConfig` is already
+    # the `[models]` block (hunter, judge, endpoint) and two meanings of one name is how you get a scan that
+    # silently reads the wrong settings. Additive -- without a CPG engine the scan is unchanged.
+    enabled: bool = True
+    # A TOTAL for the run, not a per-region cap. Conservative by default: a first scan of an unfamiliar
+    # repository should not be able to run away with someone's budget before they have decided they want it.
+    max_model_calls: int = 100
+    max_regions: int = 300
+
+
 @dataclass(frozen=True)
 class ObligationsConfig:
     # authorization-obligations: the obligation checker in MAP (standard/deep only).
@@ -168,6 +180,7 @@ class ResolvedConfig:
     regress: RegressConfig = RegressConfig()
     variants: VariantsConfig = VariantsConfig()
     obligations: ObligationsConfig = ObligationsConfig()
+    model: ModelLayerConfig = ModelLayerConfig()
     runs_dir: str = ".openultrasast/runs"
 
 
@@ -194,6 +207,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig:
         regress=_load_regress(data.get("regress", {})),
         variants=_load_variants(data.get("variants", {})),
         obligations=_load_obligations(data.get("obligations", {})),
+        model=_load_model(data.get("model", {})),
         runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"),
     )
 
@@ -369,6 +383,15 @@ def _load_variants(value: object) -> VariantsConfig:
     )
 
 
+def _load_model(value: object) -> ModelLayerConfig:
+    data = _section(value)
+    return ModelLayerConfig(
+        enabled=bool(data.get("enabled", ModelLayerConfig.enabled)),
+        max_model_calls=_int_value(data.get("max_model_calls"), ModelLayerConfig.max_model_calls),
+        max_regions=_int_value(data.get("max_regions"), ModelLayerConfig.max_regions),
+    )
+
+
 def _load_obligations(value: object) -> ObligationsConfig:
     data = _section(value)
     return ObligationsConfig(
diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py
index cb9f1aa..3a78403 100644
--- a/src/openultrasast/reports.py
+++ b/src/openultrasast/reports.py
@@ -205,6 +205,7 @@ def write_manifest(
     provenance: dict[str, object] | None = None,
     variants: dict[str, object] | None = None,
     obligations: dict[str, object] | None = None,
+    model: dict[str, object] | None = None,
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications)
@@ -246,6 +247,8 @@ def write_manifest(
         payload["variants"] = variants
     if obligations is not None:
         payload["obligations"] = obligations
+    if model is not None:
+        payload["model"] = model
     path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
 
 
diff --git a/tests/test_cli_hx_dispatch.py b/tests/test_cli_hx_dispatch.py
index 26725b2..29e2be5 100644
--- a/tests/test_cli_hx_dispatch.py
+++ b/tests/test_cli_hx_dispatch.py
@@ -41,8 +41,11 @@ def test_degradation_recorded_when_models_set_but_extra_absent(tmp_path: Path, m
 
     degradations = manifest.get("degradations", [])
     stages = {entry["stage"] for entry in degradations}
-    assert stages == {"hunter_pool", "verify"}
-    assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations)
+    # `model` is present because the model layer records that no CPG engine was available rather than being
+    # silently absent (contributor-scan Req 1.3). This test is about the HarnessX extra, so it asserts over
+    # the HarnessX stages specifically instead of the whole set.
+    assert stages == {"hunter_pool", "verify", "model"}
+    assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations if entry["stage"] != "model")
     # Fallback still produced findings (deterministic hunter ran).
     assert manifest["findings"]
 
diff --git a/tests/test_model_scan_stage.py b/tests/test_model_scan_stage.py
new file mode 100644
index 0000000..4b19ccb
--- /dev/null
+++ b/tests/test_model_scan_stage.py
@@ -0,0 +1,98 @@
+"""contributor-scan task 1.3: the model layer reaches a user (Req 1).
+
+`model-grounded-detection` finished 16/16 with the model layer having zero production callers: everything it
+measured came from a harness. This is the stage that closes that, and it is deliberately additive — the
+pattern, overlay and obligation findings all keep working, and a machine without a CPG engine scans exactly
+as it does today.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+class _Runtime:
+    """The two things `_run_model_layer` uses. A real HarnessRuntime needs a scan id, config and trace
+    writer, none of which this behaviour depends on."""
+
+    def __init__(self) -> None:
+        self.state: dict[str, list] = {"degradations": []}
+
+    def run_stage(self, name, fn):  # type: ignore[no-untyped-def]
+        return fn()
+
+
+def _repo(tmp_path: Path) -> Path:
+    (tmp_path / "app.py").write_text(
+        "import os\nfrom flask import request\n\n\ndef run():\n    cmd = request.args['c']\n    os.system(cmd)\n"
+    )
+    return tmp_path
+
+
+def test_the_model_config_block_exists_and_is_conservative() -> None:
+    """Req 8.4: optional planes default to something that cannot surprise a first-time user."""
+    from openultrasast.config import load_config
+
+    model = load_config(None).model
+    assert model.max_model_calls > 0 and model.max_regions > 0
+    assert model.max_model_calls <= 500, "a default budget should not be able to run away"
+
+
+def test_the_config_is_readable_from_toml(tmp_path: Path) -> None:
+    from openultrasast.config import load_config
+
+    path = tmp_path / "ousast.toml"
+    path.write_text("[model]\nenabled = false\nmax_model_calls = 7\nmax_regions = 3\n")
+    model = load_config(path).model
+    assert model.enabled is False and model.max_model_calls == 7 and model.max_regions == 3
+
+
+def test_without_a_cpg_engine_the_stage_degrades_and_records_why(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Req 1.3: the model layer is additive. A machine without Joern must scan exactly as it does today."""
+    from openultrasast.cli import _run_model_layer
+    from openultrasast.config import load_config
+
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off")
+    runtime = _Runtime()
+    findings, payload = _run_model_layer(
+        root=_repo(tmp_path), targets=[], entries=[], settings=load_config(None).model, runtime=runtime, config=load_config(None)
+    )
+    assert findings == []
+    assert any(d.get("reason") == "cpg_unavailable" for d in runtime.state["degradations"])
+    assert payload["skipped"] == "cpg_unavailable"
+
+
+def test_the_payload_reports_what_the_model_arbitrated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Req 1.5: a user should see how much of a result the model established, not have to infer it."""
+    from openultrasast.cli import _run_model_layer
+    from openultrasast.config import load_config
+
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off")
+    _, payload = _run_model_layer(
+        root=_repo(tmp_path), targets=[], entries=[], settings=load_config(None).model, runtime=_Runtime(), config=load_config(None)
+    )
+    assert set(payload) >= {"by_rung", "regions_scanned", "regions_unjudged", "model_calls", "cost_usd"}
+
+
+def test_unjudged_regions_are_reported_with_their_rank(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """What the budget could not reach is what a large-repository exclusion list should be built from, so the
+    lowest-ranked skipped regions are named rather than merely counted."""
+    from openultrasast.cli import _model_payload
+    from openultrasast.model.scan import ModelScanResult
+
+    result = ModelScanResult(by_rung={}, regions_scanned=2, regions_unjudged=8, model_calls=2, cost_usd=0.01)
+    payload = _model_payload(result, unjudged_paths=("low/a.py", "low/b.py"))
+    assert payload["regions_unjudged"] == 8
+    assert payload["unjudged_sample"] == ["low/a.py", "low/b.py"]
+
+
+def test_a_disabled_model_layer_runs_nothing(tmp_path: Path) -> None:
+    from openultrasast.cli import _run_model_layer
+    from openultrasast.config import ModelLayerConfig, load_config
+
+    findings, payload = _run_model_layer(
+        root=_repo(tmp_path), targets=[], entries=[], settings=ModelLayerConfig(enabled=False), runtime=_Runtime(), config=load_config(None)
+    )
+    assert findings == [] and payload["skipped"] == "disabled"
diff --git a/tests/test_scan_pipeline_integrity.py b/tests/test_scan_pipeline_integrity.py
index 2a3ade5..45d1238 100644
--- a/tests/test_scan_pipeline_integrity.py
+++ b/tests/test_scan_pipeline_integrity.py
@@ -65,10 +65,14 @@ def test_standard_scan_extra_absent_matches_deterministic_baseline(tmp_path: Pat
 
     assert requested == baseline  # byte-identical findings -> sync driver behaviour unchanged
     stages = {entry["stage"] for entry in requested_manifest.get("degradations", [])}
-    assert stages == {"hunter_pool", "verify"}
-    assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in requested_manifest["degradations"])
+    # `model` joins the set when the model layer runs without a CPG engine: findings are byte-identical
+    # (asserted above) and the only difference is that the scan now SAYS the arbiter was unavailable
+    # (contributor-scan Req 1.3) rather than being silently absent.
+    assert stages == {"hunter_pool", "verify", "model"}
+    assert {e["reason"] for e in requested_manifest["degradations"] if e["stage"] == "model"} == {"cpg_unavailable"}
+    assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in requested_manifest["degradations"] if entry["stage"] != "model")
     assert all(entry["reason"] != "hunter_model_unavailable" for entry in requested_manifest["degradations"])
-    assert [entry["reason"] for entry in baseline_manifest.get("degradations", [])] == ["hunter_model_unavailable"]
+    assert [e["reason"] for e in baseline_manifest.get("degradations", []) if e["stage"] != "model"] == ["hunter_model_unavailable"]
 
 
 def test_quick_scan_is_unaffected_by_the_extra(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: