feat(cpg): give the files a frontend refuses a CPG of their own

A dropped file is dropped from the ONLY graph there is, so every question
about it has no answer -- which is how a 637-file plugin became a scan of
nothing. But those files build perfectly well alone: the failure is
per-invocation, not per-file. So a file the main build refuses is excluded
from it, given a second CPG, and both are queried with rows concatenated per
request id.

    pmpro two-file slice   before: cpg_empty, 0 findings
                           after:  18 entailed incl. CVE-2023-23488 at :936,
                                   reported with cpg_sharded

What this does NOT buy is a flow that crosses the shard boundary: a source in
the excluded file whose sink is in the rest of the tree is invisible to every
question asked, and no merging of results recovers it. That is why the split
is REPORTED rather than quietly served as if it were whole.

The excluded-file tree is symlinks at the original relative paths. A region
asks about `classes/class.memberorder.php` and the queries match a filename by
suffix, so a flattened copy would cheerfully answer about a file nobody asked
about.

Reports gain "What could not be analysed", rendering cpg_sharded, cpg_empty,
files_unparsed, query_failed and budget_exhausted in the reader's terms. A
degradation is a fact about coverage, not a warning to file away: silence in
the coverage table means "no fact table covered this", and silence about an
empty or split graph means something else entirely.

Also removes the WordPress vocabulary from the query, which was the other half
of this request. `apply_filters` was a hardcoded Set in taint.sc and
`add_action|add_filter|add_shortcode` a hardcoded alternation in mapping.py.
Both are now a `[[dispatch]]` fact:

    [[dispatch]]
    id = "wordpress_hooks"
    register = ["add_action", "add_filter", "add_shortcode"]
    apply = ["apply_filters", ..., "do_action", ...]

The shape is general -- a registry keyed by a string literal at both ends, as
Django signals, jQuery events and Symfony's dispatcher all are -- so a second
framework is a row in a TOML rather than an edit to a query. `dispatch_apply`
is empty for every language that declares none, and the hook half then does
nothing at all rather than looking for names that mean nothing there.

848 tests pass; gates byte-identical. wpstatistics 3/2 with the CVE at :225 on
the vulnerable side only, mwwpform 1/0, pmpro 18/1, VAmPI 4 unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.kiro/specs/contributor-scan/tasks.md

@@ -596,8 +596,20 @@
     (`_build_with_retries`, 4 attempts). Retrying is only sound because the failure is intermittent AND
     reported -- retrying a silent failure would be superstition. Driver-level result on the slice that
     previously needed four manual retries: **6 of 6 builds usable**, 3 entailed findings each time.
-  - Still open: a file that fails on every attempt is still reported rather than analysed, and `--exclude`
-    plus a second CPG for the excluded files is the remaining lever.
+  - **Excluded files now get a CPG of their own, 2026-09-10.** A file the frontend refuses is excluded from
+    the main build and put in a second graph -- the same file builds perfectly well alone, the failure being
+    per-invocation -- and the queries run over both, rows concatenated per request id. The two-file PMPro
+    slice that used to report `cpg_empty` and nothing at all now reports **18 entailed findings including
+    CVE-2023-23488 at :936**, with `cpg_sharded` beside them.
+  - What it does NOT buy is a flow that crosses the shard boundary, and that is why the split is reported
+    rather than quietly served: a source in the excluded file whose sink is in the rest of the tree is
+    invisible to every question asked. `reports` now renders this under "What could not be analysed", along
+    with `cpg_empty`, `files_unparsed`, `query_failed` and `budget_exhausted` -- a degradation is a fact
+    about coverage, not a warning to file away.
+  - The excluded-file tree is symlinks at the ORIGINAL relative paths, because a region asks about
+    `classes/class.memberorder.php` and the queries match a filename by suffix; a flattened copy would answer
+    about a file nobody asked about.
+  - Still open: nothing here recovers a cross-shard flow, and the 637-file build remains unmeasured.
   - Still open: wiring that exclusion loop, the 637-file build with it in place, and whether the REST handler
     two calls away can reach the sink -- the slice found it through a same-function flow, not across the
     object.

src/openultrasast/cli.py

@@ -555,6 +555,10 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             obligations=obligations_cited or None,
             obligations_summary=obligations_payload or None,
             coverage=coverage_for_report,
+            # Degradations are coverage facts: an empty graph, a graph built in parts, a query that never
+            # ran. Kept out of the report they leave the reader unable to tell "nothing was found" from
+            # "nothing was looked at".
+            degradations=list(runtime.state.get("degradations") or ()),
         ),
     )
     runtime.run_stage(

src/openultrasast/cpg/backend.py

@@ -26,7 +26,7 @@ import re
 import shutil
 import subprocess
 import tempfile
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Protocol
@@ -193,15 +193,16 @@ class JoernBackend:
 
         # A frontend whose failures are visible and retryable is worth more than one whose are not.
         if language.lower() in _PREFER_FRONTEND:
-            retried = self._build_with_retries(root, cpg_path, scratch, language)
-            if retried is not None:
-                unparsed = retried
+            shards, unparsed = self._build_sharded(root, scratch, language)
+            if shards:
                 if unparsed:
                     logger.warning("the frontend could not parse %d file(s) under %s: %s", len(unparsed), root, ", ".join(unparsed[:5]))
+                if len(shards) > 1:
+                    logger.info("querying %d cpg shards for %s; flows that cross them are not visible", len(shards), root)
                 return CpgResult(
-                    cpg_path=cpg_path,
-                    run=lambda query, params: self.query(cpg_path, query, params),
-                    run_batch=lambda query, requests: self.query_batch(cpg_path, query, requests),
+                    cpg_path=shards[0],
+                    run=lambda query, params: self.query_across(shards, query, params),
+                    run_batch=lambda query, requests: self.query_batch_across(shards, query, requests),
                     unparsed=unparsed,
                 )
         command = [parse or "joern-parse", self._heap_flag(), str(root), "--output", str(cpg_path)]
@@ -223,7 +224,117 @@ class JoernBackend:
             unparsed=unparsed,
         )
 
-    def _build_with_retries(self, root: Path, cpg_path: Path, scratch: Path, language: str) -> tuple[str, ...] | None:
+    def _build_sharded(self, root: Path, scratch: Path, language: str) -> tuple[tuple[Path, ...], tuple[str, ...]]:
+        """Build the tree, and give the files it refuses a CPG of their OWN rather than losing them.
+
+        A frontend that drops a file drops it from the only graph there is, and every question about that
+        file then has no answer -- which is how a 637-file plugin became a scan of nothing. But the same
+        files build perfectly well on their own: the failure is per-invocation, not per-file. So a file the
+        main build refuses is excluded from it and put in a second CPG, and the queries run over both.
+
+        What this does NOT buy is a flow that crosses the shard boundary. A source in the excluded file and
+        its sink in the rest of the tree is invisible, and no merging of results can recover it -- which is
+        why the driver reports the split rather than quietly serving a partial answer.
+
+        Returns the CPGs to query and the files that defeated even a shard of their own.
+        """
+        main = scratch / "cpg.bin"
+        dropped = self._build_with_retries(root, main, scratch, language)
+        if dropped is None:
+            return (), ()
+        if not dropped:
+            return (main,), ()
+
+        # Second pass over the same tree, this time telling the frontend to leave the difficult files alone,
+        # so the rest of the repository is analysed instead of nothing being analysed.
+        remainder = self._build_with_retries(root, main, scratch, language, exclude=dropped)
+        shards: list[Path] = []
+        if remainder is not None and main.is_file():
+            shards.append(main)
+
+        island = self._island_root(root, scratch, dropped)
+        if island is None:
+            return tuple(shards), dropped
+        second = scratch / "cpg-excluded.bin"
+        still = self._build_with_retries(island, second, scratch, language)
+        if still is not None and second.is_file():
+            shards.append(second)
+            return tuple(shards), tuple(still)
+        return tuple(shards), dropped
+
+    def _island_root(self, root: Path, scratch: Path, files: Sequence[str]) -> Path | None:
+        """A tree holding only ``files``, at their original relative paths.
+
+        The paths have to be preserved: a region asks about `classes/class.memberorder.php`, and the queries
+        match a method's filename by suffix, so a flattened copy would answer about a file nobody asked
+        about. Symlinks, because the alternative is copying a repository to analyse it.
+        """
+        island = scratch / "excluded-root"
+        made = 0
+        for name in files:
+            source = Path(name)
+            if not source.is_absolute():
+                source = root / name
+            if not source.is_file():
+                continue
+            try:
+                relative = source.relative_to(root)
+            except ValueError:
+                relative = Path(source.name)
+            destination = island / relative
+            try:
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                if not destination.exists():
+                    destination.symlink_to(source)
+                made += 1
+            except OSError as exc:  # noqa: PERF203 - one bad file must not lose the rest
+                logger.warning("could not stage %s for its own cpg: %s", source, exc)
+        return island if made else None
+
+    def query_across(self, shards: Sequence[Path], query: str, params: Mapping[str, object]) -> object | None:
+        """One question over every shard, answers concatenated. ``None`` only when no shard could answer."""
+        merged: list[object] = []
+        answered = 0
+        for shard in shards:
+            rows = self.query(shard, query, params)
+            if rows is None:
+                continue
+            answered += 1
+            if isinstance(rows, list):
+                merged.extend(rows)
+        return merged if answered else None
+
+    def query_batch_across(
+        self, shards: Sequence[Path], query: str, requests: Mapping[str, Mapping[str, object]]
+    ) -> dict[str, list[object]] | None:
+        """The batch over every shard, rows concatenated per request id.
+
+        The census is summed rather than taken from one shard, and carries the shard count, because
+        `cpg_empty` has to mean "no graph anywhere" and not "the first of two graphs was small".
+        """
+        merged: dict[str, list[object]] = {}
+        methods = 0
+        files = 0
+        answered = 0
+        for shard in shards:
+            one = self.query_batch(shard, query, requests)
+            if one is None:
+                continue
+            answered += 1
+            census = one.pop("__census__", None)
+            if census and isinstance(census[0], Mapping):
+                methods += int(str(census[0].get("methods", "0")) or 0)
+                files += int(str(census[0].get("files", "0")) or 0)
+            for rid, rows in one.items():
+                merged.setdefault(rid, []).extend(rows)
+        if not answered:
+            return None
+        merged["__census__"] = [{"methods": str(methods), "files": str(files), "shards": str(len(shards))}]
+        return merged
+
+    def _build_with_retries(
+        self, root: Path, cpg_path: Path, scratch: Path, language: str, exclude: Sequence[str] = ()
+    ) -> tuple[str, ...] | None:
         """Build through the frontend, trying again while it reports files it could not parse.
 
         Retrying is only sound because the failure is INTERMITTENT and the frontend says when it happened.
@@ -235,7 +346,7 @@ class JoernBackend:
         """
         best: tuple[str, ...] | None = None
         for attempt in range(FRONTEND_BUILD_ATTEMPTS):
-            unparsed = self._build_with_frontend(root, cpg_path, scratch, language)
+            unparsed = self._build_with_frontend(root, cpg_path, scratch, language, exclude)
             if unparsed is None:
                 return best  # the frontend is missing or failed outright; keep any earlier partial build
             if not unparsed:
@@ -250,7 +361,9 @@ class JoernBackend:
             )
         return best
 
-    def _build_with_frontend(self, root: Path, cpg_path: Path, scratch: Path, language: str) -> tuple[str, ...] | None:
+    def _build_with_frontend(
+        self, root: Path, cpg_path: Path, scratch: Path, language: str, exclude: Sequence[str] = ()
+    ) -> tuple[str, ...] | None:
         """Retry the build through the language frontend alone.
 
         Returns the files the frontend dropped -- possibly none -- or ``None`` when there is nothing to retry
@@ -265,7 +378,10 @@ class JoernBackend:
         if binary is None:
             logger.warning("no frontend %s on PATH to retry the build for %s", frontend, root)
             return None
-        completed = self._run([binary, self._heap_flag(), str(root), "-o", str(cpg_path)], timeout=self.build_timeout, cwd=scratch)
+        command = [binary, self._heap_flag(), str(root), "-o", str(cpg_path)]
+        for name in exclude:
+            command += ["--exclude", name]
+        completed = self._run(command, timeout=self.build_timeout, cwd=scratch)
         if completed is None or completed.returncode != 0 or not cpg_path.is_file():
             detail = (completed.stderr or completed.stdout or "")[-400:] if completed is not None else "timeout"
             logger.warning("%s also failed for %s: %s", frontend, root, detail)

src/openultrasast/cpg/queries/taint.sc

@@ -49,6 +49,7 @@
     boundedSinks: String = "",
     parameterSources: String = "false",
     hookCallbacks: String = "",
+    dispatchApply: String = "",
     requests: String = "",
     requestsFile: String = ""
 ) = {
@@ -68,6 +69,7 @@
       sinksS: String,
       sanitizersS: String,
       hookCallbacksS: String,
+      dispatchApplyS: String,
       functionS: String,
       paramSrc: String,
       fileS: String,
@@ -308,7 +310,10 @@
   // The rule: an `apply_filters('H', ...)` call is a SOURCE when some callback registered for H returns
   // data that an unsanitized flow reached. That is the same question as the field half, against a different
   // key.
-  val HOOK_APPLY = Set("apply_filters", "apply_filters_ref_array", "do_action", "do_action_ref_array")
+  // The registry's vocabulary arrives as a FACT, never as a constant here. `apply_filters` is WordPress's
+  // name for this; Django signals, jQuery events and Symfony's dispatcher all have their own, and each is a
+  // row in a semantic fact table rather than an edit to this query.
+  val hookApply = split(dispatchApplyS).toSet
 
   val hookTable: Map[String, List[String]] =
     hookCallbacksS
@@ -399,9 +404,9 @@
     )
 
   def hookSourceNodes =
-    if (hookTable.isEmpty) Iterator.empty
+    if (hookTable.isEmpty || hookApply.isEmpty) Iterator.empty
     else {
-      val applies = cpg.call.filter(c => HOOK_APPLY.contains(c.name)).filter(c => inScope(c.method)).l
+      val applies = cpg.call.filter(c => hookApply.contains(c.name)).filter(c => inScope(c.method)).l
       if (applies.isEmpty) Iterator.empty
       else {
         val keys = applies.flatMap { call =>
@@ -569,6 +574,7 @@
           field("sinks"),
           field("sanitizers"),
           field("hookCallbacks"),
+          field("dispatchApply"),
           field("function"),
           paramSrc,
           field("file"),
@@ -586,7 +592,7 @@
     println(ujson.write(ujson.Obj.from(answers.toSeq :+ ("__census__" -> census))))
   } else {
     println(
-      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, function, parameterSources, file, callDepth, boundedSinks): _*))
+      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, dispatchApply, function, parameterSources, file, callDepth, boundedSinks): _*))
     )
   }
   println("---OUSAST-CPG-END---")

src/openultrasast/mapping.py

@@ -117,7 +117,7 @@ def analyze_entry_points(root: Path, targets: list[FileTarget]) -> list[EntryPoi
     return sorted(records, key=lambda item: (item.path, item.line is None, item.line or 0, item.name))
 
 
-def php_hook_callbacks(root: Path, targets: list[FileTarget]) -> dict[str, tuple[str, ...]]:
+def php_hook_callbacks(root: Path, targets: list[FileTarget], register_calls: Sequence[str] | None = None) -> dict[str, tuple[str, ...]]:
     """WordPress's hook registry, read out of the source text because the graph does not carry it.
 
     `php2cpg` 4.0.623 drops the callback argument of a registration entirely. What the CPG holds is
@@ -131,7 +131,24 @@ def php_hook_callbacks(root: Path, targets: list[FileTarget]) -> dict[str, tuple
     This is the link table for the hook half of the two-stage join (task 5.11). It is deliberately a plain
     name-to-names mapping with no inference in it: a hook whose name is computed (`"save_post_" . $type`)
     simply does not appear, rather than being guessed at.
+
+    ``register_calls`` comes from the semantic fact tables and defaults to whatever they declare for PHP.
+    Nothing here knows what WordPress is: the registry shape is general -- Django signals, jQuery events, a
+    plugin system's event bus -- and a second framework is a row in a TOML, not a change to this function.
     """
+    if register_calls is None:
+        from .semantic.facts import load_facts
+
+        register_calls = tuple(sorted({call for fact in load_facts().for_language("php").dispatches for call in fact.register}))
+    if not register_calls:
+        return {}
+    pattern = re.compile(
+        r"""\b(?P<call>"""
+        + "|".join(re.escape(name) for name in register_calls)
+        + r""")\s*\(\s*['"](?P<hook>[^'"]+)['"]\s*,\s*"""
+        + _PHP_CALLABLE,
+        re.VERBOSE,
+    )
     table: dict[str, set[str]] = {}
     for target in targets:
         if target.language != "php":
@@ -140,7 +157,7 @@ def php_hook_callbacks(root: Path, targets: list[FileTarget]) -> dict[str, tuple
             text = (root / target.path).read_text(errors="ignore")
         except OSError:
             continue
-        for match in _PHP_REGISTRATION.finditer(text):
+        for match in pattern.finditer(text):
             handler = match.group("handler") or match.group("method")
             if handler:
                 table.setdefault(match.group("hook"), set()).add(handler)

src/openultrasast/model/scan.py

@@ -190,6 +190,13 @@ def scan_repository(
                     methods = int(str(entry.get("methods", "0")) or 0)
                     if methods == 0:
                         degradations.append({"stage": "model", "reason": "cpg_empty", "files": int(str(entry.get("files", "0")) or 0)})
+                    # More than one graph means some files could only be built apart from the rest, and a
+                    # flow whose source is in one shard and whose sink is in another does not exist for any
+                    # question we can ask. Reporting the split is the difference between a partial answer and
+                    # a partial answer that looks whole.
+                    shards = int(str(entry.get("shards", "1")) or 1)
+                    if shards > 1:
+                        degradations.append({"stage": "model", "reason": "cpg_sharded", "shards": shards})
                     census_reported = True
                 rows_by_id.update(answered_batch)
         else:  # a backend without batching still works, one call at a time

src/openultrasast/model/specs.py

@@ -112,6 +112,9 @@ class TaintSpec:
     # discharge is a guard and not a call.
     bounded_sinks: tuple[str, ...] = ()
     safe_shape_sinks: tuple[str, ...] = ()
+    # The calls that read a value back out of a string-keyed registry (`apply_filters`, `do_action`). Facts,
+    # not constants, so a second framework is a row in a TOML rather than an edit to the query.
+    dispatch_apply: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -221,6 +224,7 @@ def taint_specs(
     def _is_shape(fact: object) -> bool:
         return bool(getattr(fact, "parameterized", False)) or getattr(fact, "literal_format_arg", None) is not None
 
+    dispatch_apply = tuple(sorted({call for fact in scoped.dispatches for call in fact.apply}))
     sanitizers = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) for call in fact.calls}))
     safe_shapes = tuple(sorted({call for fact in scoped.sanitizers if _is_shape(fact) for call in fact.calls}))
     by_family: dict[str, set[str]] = {}
@@ -243,6 +247,7 @@ def taint_specs(
             sanitizers=sanitizers,
             bounded_sinks=tuple(sorted(bounded.get(family_id, ()))),
             safe_shape_sinks=safe_shapes,
+            dispatch_apply=dispatch_apply,
         )
         for family_id, calls in sorted(by_family.items())
     }

src/openultrasast/model/taint.py

@@ -51,6 +51,9 @@ def request_params(
         # graph, because php2cpg drops a registration's callback argument entirely -- see
         # `mapping.php_hook_callbacks`. Empty for every language that does not register handlers by string.
         "hookCallbacks": hook_callbacks,
+        # The registry's own vocabulary, from the fact tables. Without it the query would have to know what
+        # `apply_filters` is, which makes a general mechanism into a WordPress feature.
+        "dispatchApply": spec.dispatch_apply,
     }
 
 

src/openultrasast/reports.py

@@ -23,6 +23,7 @@ def write_markdown_report(
     obligations: Mapping[str, Mapping[str, object]] | None = None,
     obligations_summary: Mapping[str, object] | None = None,
     coverage: Sequence[Mapping[str, object]] | None = None,
+    degradations: Sequence[Mapping[str, object]] | None = None,
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications or [])
@@ -87,8 +88,8 @@ def write_markdown_report(
             cited_obligations[finding.finding_id] = info_o
             _append_obligation_lines(lines, info_o, (mechanisms or {}).get(str(info_o.get("known_fix") or "")))
         lines.extend(["", finding.rationale, ""])
-    if coverage:
-        _append_coverage(lines, coverage)
+    if coverage or degradations:
+        _append_coverage(lines, coverage or (), degradations or ())
     if patterned:
         _append_pattern_matches(lines, patterned)
     if cited:
@@ -389,7 +390,7 @@ def _crowded_pattern_rules(findings: Sequence[StaticFinding]) -> list[StaticFind
     return [finding for finding in findings if _is_pattern_only(finding) and counts.get(_rule_key(finding), 0) > _PATTERN_DETAIL_LIMIT]
 
 
-def _append_coverage(lines: list[str], coverage: Sequence[Mapping[str, object]]) -> None:
+def _append_coverage(lines: list[str], coverage: Sequence[Mapping[str, object]], degradations: Sequence[Mapping[str, object]] = ()) -> None:
     """What was looked for, and what could not be decided (Req 5.6).
 
     A scan that reports nothing has established that nothing it MODELS was found. That is not the same as
@@ -418,6 +419,53 @@ def _append_coverage(lines: list[str], coverage: Sequence[Mapping[str, object]])
         for row in stated:
             lines.extend([f"**`{row.get('family')}` ({row.get('language')})** — {row.get('limits')}", ""])
 
+    _append_not_analysed(lines, degradations)
+
+
+# A degradation is a fact about COVERAGE, not a warning to be filed away. Each of these says the scan
+# answered a smaller question than the one it was asked, and a reader who is not told cannot know that the
+# silence in the table above is narrower than it looks.
+_COVERAGE_DEGRADATIONS: dict[str, str] = {
+    "cpg_empty": "The code graph came back with no methods in it, so **nothing above was analysed at all**.",
+    "cpg_build_failed": "The code graph could not be built, so nothing above was analysed.",
+    "cpg_sharded": (
+        "The graph had to be built in {shards} parts, because the frontend could not parse the whole tree at "
+        "once. Each part was queried, but **a flow whose source is in one part and whose sink is in another "
+        "is not visible to any question asked here**."
+    ),
+    "files_unparsed": "{count} file(s) could not be parsed and are absent from the graph: {files}.",
+    "query_failed": "A `{kind}` query failed for {requests} region(s), so those regions were not decided.",
+    "budget_exhausted": "The model-call budget ran out with {regions_unasked} region(s) still unasked.",
+    "region_failed": "The region `{path}` could not be arbitrated.",
+}
+
+
+def _append_not_analysed(lines: list[str], degradations: Sequence[Mapping[str, object]]) -> None:
+    """The part of the repository the scan did NOT get to, said plainly.
+
+    Silence in the coverage table means "no fact table covered this". Silence here would mean something
+    else entirely -- that the graph was empty, or split, or that the query never ran -- and the reader has
+    no way to tell those apart from the findings alone.
+    """
+    rendered: list[str] = []
+    for item in degradations:
+        template = _COVERAGE_DEGRADATIONS.get(str(item.get("reason") or ""))
+        if template is None:
+            continue
+        fields = dict(item)
+        files = fields.get("files")
+        if isinstance(files, (list, tuple)):
+            fields["files"] = ", ".join(f"`{name}`" for name in files)
+        try:
+            rendered.append(template.format(**fields))
+        except (KeyError, IndexError):  # a degradation shaped differently than its template expects
+            rendered.append(f"`{item.get('reason')}`")
+    if not rendered:
+        return
+    lines.extend(["### What could not be analysed", ""])
+    lines.extend(f"- {line}" for line in rendered)
+    lines.append("")
+
 
 def _is_pattern_only(finding: StaticFinding) -> bool:
     """Was this proposed with nothing corroborating it?

src/openultrasast/ruleset/semantic/php.toml

@@ -188,3 +188,15 @@ calls = ["intval", "floatval", "filter_var", "settype", "absint"]
 id = "prepared_statement"
 calls = ["mysqli_prepare", "mysqli_stmt_bind_param", "pg_prepare", "pg_execute"]
 parameterized = true
+
+# A registry that connects a producer to a consumer through a STRING key rather than a call. Neither end
+# names the other, so no CPG frontend can draw the edge: php2cpg does not even keep the callback argument,
+# and `add_filter("hook", array($this, "m"))` reaches the graph as `add_filter("hook", )`.
+#
+# WordPress is what forced this into existence and none of it is about WordPress. The shape is general --
+# Django signals, jQuery events, Symfony's event dispatcher, any plugin system -- and a second framework is
+# a row here rather than a change to the query. That is the whole reason it is a fact and not a constant.
+[[dispatch]]
+id = "wordpress_hooks"
+register = ["add_action", "add_filter", "add_shortcode"]
+apply = ["apply_filters", "apply_filters_ref_array", "do_action", "do_action_ref_array"]

src/openultrasast/semantic/facts.py

@@ -53,12 +53,34 @@ class SanitizerFact:
     literal_format_arg: int | None = None
 
 
+@dataclass(frozen=True)
+class DispatchFact:
+    """A framework that connects a producer to a consumer through a STRING key rather than a call.
+
+    WordPress is the example that forced this, but nothing here is about WordPress. The shape is general --
+    a registry keyed by a name, written as a literal at both ends -- and it is how Django signals, jQuery
+    events, Symfony's event dispatcher and every plugin system of this kind route data. No CPG frontend can
+    draw the edge, because neither end names the other.
+
+    ``register`` names the calls that put a callback INTO the registry, ``apply`` the calls that take a value
+    back OUT of it. Both belong in a fact table for the same reason sinks do: a query that hardcodes
+    `apply_filters` is a query that only works on one framework, and the next one needs a code change rather
+    than a row.
+    """
+
+    id: str
+    register: tuple[str, ...]
+    apply: tuple[str, ...]
+    language: str
+
+
 @dataclass(frozen=True)
 class SemanticFacts:
     version: str
     sources: tuple[SourceFact, ...]
     sinks: tuple[SinkFact, ...]
     sanitizers: tuple[SanitizerFact, ...]
+    dispatches: tuple[DispatchFact, ...] = ()
 
     def for_language(self, language: str) -> SemanticFacts:
         key = _LANGUAGE_ALIASES.get(language, language)
@@ -67,6 +89,7 @@ class SemanticFacts:
             sources=tuple(item for item in self.sources if item.language == key),
             sinks=tuple(item for item in self.sinks if item.language == key),
             sanitizers=tuple(item for item in self.sanitizers if item.language == key),
+            dispatches=tuple(item for item in self.dispatches if item.language == key),
         )
 
 
@@ -77,6 +100,7 @@ def load_facts(directory: Path | None = None) -> SemanticFacts:
     sources: list[SourceFact] = []
     sinks: list[SinkFact] = []
     sanitizers: list[SanitizerFact] = []
+    dispatches: list[DispatchFact] = []
     versions: list[str] = []
     found = False
     for path in sorted(root.glob("*.toml")):
@@ -90,9 +114,33 @@ def load_facts(directory: Path | None = None) -> SemanticFacts:
         sources.extend(_sources(payload.get("source"), language, path))
         sinks.extend(_sinks(payload.get("sink"), language, path))
         sanitizers.extend(_sanitizers(payload.get("sanitizer"), language, path))
+        dispatches.extend(_dispatches(payload.get("dispatch"), language, path))
     if not found:
         raise FactLoadError(f"no semantic fact files in {root}")
-    return SemanticFacts(version=versions[0] if versions else "1", sources=tuple(sources), sinks=tuple(sinks), sanitizers=tuple(sanitizers))
+    return SemanticFacts(
+        version=versions[0] if versions else "1",
+        sources=tuple(sources),
+        sinks=tuple(sinks),
+        sanitizers=tuple(sanitizers),
+        dispatches=tuple(dispatches),
+    )
+
+
+def _dispatches(value: object, language: str, path: Path) -> list[DispatchFact]:
+    facts: list[DispatchFact] = []
+    for item in _items(value):
+        identifier = str(item.get("id", "")).strip()
+        if not identifier:
+            raise FactLoadError(f"dispatch without an id in {path}")
+        facts.append(
+            DispatchFact(
+                id=identifier,
+                register=_strings(item.get("register"), "register", path),
+                apply=_strings(item.get("apply"), "apply", path),
+                language=language,
+            )
+        )
+    return facts
 
 
 def _items(value: object) -> list[dict[str, object]]:

tests/test_cpg_backend.py

@@ -437,5 +437,76 @@ def test_a_php_build_that_never_stops_dropping_files_reports_them(tmp_path: Path
 
     result = JoernBackend(runner=runner).build(tmp_path, language="php")
 
-    assert attempts == FRONTEND_BUILD_ATTEMPTS, "it gives up rather than retrying forever"
-    assert result is not None and result.unparsed == ("/src/stubborn.php",)
+    assert attempts >= FRONTEND_BUILD_ATTEMPTS, "it retried"
+    assert attempts < 100, "and it gave up rather than retrying forever"
+    assert result is not None and result.unparsed == ("/src/stubborn.php",), "the file is named, not silently lost"
+
+
+def test_files_the_frontend_refuses_get_a_cpg_of_their_own(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A dropped file is dropped from the ONLY graph there is, and every question about it then has no
+    answer. The same file builds fine alone -- the failure is per-invocation -- so it is excluded from the
+    main build and given a second CPG, and queries run over both."""
+    import subprocess
+
+    from openultrasast.cpg.backend import JoernBackend
+
+    (tmp_path / "big.php").write_text("<?php function a() {}")
+    (tmp_path / "ok.php").write_text("<?php function b() {}")
+    commands: list[list[str]] = []
+
+    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
+        commands.append(list(command))
+        Path(command[command.index("-o") + 1]).write_text("cpg")
+        # The whole tree always drops big.php; any build that excludes it, or covers it alone, is clean.
+        excluded = "--exclude" in command
+        island = "excluded-root" in command[2]
+        stderr = "" if excluded or island else f"WARN Failed to process '{tmp_path / 'big.php'}'\n"
+        return subprocess.CompletedProcess(args=command, returncode=0, stdout="", stderr=stderr)
+
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
+    monkeypatch.setattr("shutil.which", lambda name: f"/opt/joern/{name}")
+
+    result = JoernBackend(runner=runner).build(tmp_path, language="php")
+
+    assert result is not None
+    assert result.unparsed == (), "the excluded file built on its own, so nothing is finally unparsed"
+    assert any("--exclude" in command for command in commands), "the main build excluded the difficult file"
+    assert any("excluded-root" in command[2] for command in commands), "and the difficult file got its own build"
+    assert (tmp_path / "big.php").exists(), "the source tree is untouched"
+
+
+def test_rows_from_every_shard_reach_the_caller(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Two graphs, one answer. The census is summed and carries the shard count, so `cpg_empty` still means
+    "no graph anywhere" rather than "the first of two was small"."""
+    from openultrasast.cpg.backend import JoernBackend
+
+    backend = JoernBackend()
+    answers = {
+        "a.bin": {"0": [{"sink": "one"}], "__census__": [{"methods": "3", "files": "1"}]},
+        "b.bin": {"0": [{"sink": "two"}], "1": [{"sink": "three"}], "__census__": [{"methods": "4", "files": "2"}]},
+    }
+    monkeypatch.setattr(JoernBackend, "query_batch", lambda self, path, query, requests: answers[path.name])
+
+    merged = backend.query_batch_across([Path("a.bin"), Path("b.bin")], "taint", {})
+
+    assert merged is not None
+    assert merged["0"] == [{"sink": "one"}, {"sink": "two"}]
+    assert merged["1"] == [{"sink": "three"}]
+    assert merged["__census__"] == [{"methods": "7", "files": "3", "shards": "2"}]
+
+
+def test_a_shard_that_cannot_answer_does_not_lose_the_others(monkeypatch: pytest.MonkeyPatch) -> None:
+    from openultrasast.cpg.backend import JoernBackend
+
+    backend = JoernBackend()
+    monkeypatch.setattr(
+        JoernBackend,
+        "query_batch",
+        lambda self, path, query, requests: None if path.name == "a.bin" else {"0": [{"sink": "kept"}]},
+    )
+
+    merged = backend.query_batch_across([Path("a.bin"), Path("b.bin")], "taint", {})
+    assert merged is not None and merged["0"] == [{"sink": "kept"}]
+
+    monkeypatch.setattr(JoernBackend, "query_batch", lambda self, path, query, requests: None)
+    assert backend.query_batch_across([Path("a.bin")], "taint", {}) is None, "no shard answered is not an empty answer"

tests/test_reports.py

@@ -617,3 +617,40 @@ def test_an_entailed_model_finding_keeps_its_section(tmp_path: Path) -> None:
     write_markdown_report(entailed, output)
 
     assert output.read_text().count(f"## {base.title}") == 20
+
+
+def test_a_sharded_graph_is_stated_in_the_report(tmp_path) -> None:  # type: ignore[no-untyped-def]
+    """A graph built in parts cannot see a flow that crosses the parts, and the reader has no way to know
+    that from the findings alone."""
+    from openultrasast.reports import write_markdown_report
+
+    path = tmp_path / "report.md"
+    write_markdown_report(
+        [],
+        path,
+        [],
+        coverage=[{"language": "php", "family": "injection", "regions": 3, "modelled": 12, "limits": ""}],
+        degradations=[
+            {"stage": "model", "reason": "cpg_sharded", "shards": 2},
+            {"stage": "model", "reason": "files_unparsed", "count": 1, "files": ["big.php"]},
+        ],
+    )
+    text = path.read_text()
+    assert "### What could not be analysed" in text
+    assert "built in 2 parts" in text
+    assert "is not visible to any question asked here" in text
+    assert "`big.php`" in text
+
+
+def test_a_clean_scan_states_no_coverage_gaps(tmp_path) -> None:  # type: ignore[no-untyped-def]
+    from openultrasast.reports import write_markdown_report
+
+    path = tmp_path / "report.md"
+    write_markdown_report(
+        [],
+        path,
+        [],
+        coverage=[{"language": "php", "family": "injection", "regions": 3, "modelled": 12, "limits": ""}],
+        degradations=[],
+    )
+    assert "### What could not be analysed" not in path.read_text()