feat(ingestion): mint Destination nodes from AsyncAPI 3.x documents (#3140)

* feat(ingestion): read AsyncAPI 3.x documents into broker addresses

Adds a format-driven reader that turns the `operations[]` entries of an
AsyncAPI 3.x document into (broker, address, direction) triples, plus the
protocol-to-broker map behind it. Nothing consumes it yet.

The reader lives outside `frameworks/spring/` on purpose, like
`destination-key.ts` and for the same reason: an AsyncAPI document is a
published artifact emitted by generators across several language toolchains
and written by hand as often as generated. The entry criterion is therefore
the document format -- a root `asyncapi` key -- and never the generator.

AsyncAPI 2.x is refused under its own countable reason rather than mapped.
Its `publish`/`subscribe` are inverted relative to 3.x `send`/`receive`, so
a naive mapping reverses every direction in the async graph while leaving it
connected: nothing fails, the arrows simply point the wrong way. A silent
skip would be indistinguishable from "this service publishes no document",
which is the one thing the refusal count has to be able to tell us.

The broker is read twice over -- from the operation's bindings and from its
channel's server protocol -- and the two readings must agree. A destination
keyed on the wrong broker joins a stranger, and with the document
contradicting itself there is no way to tell which reading is right, so the
operation is refused rather than decided by a coin flip.

An unmapped protocol passes through as its own literal instead of being
dropped, because `destinationNodeKey` takes a plain string precisely so a
non-Spring caller can attest to a broker Spring has no member for.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(cli): add --asyncapi-spec, an explicit path to AsyncAPI documents

Threads an `asyncApiSpecPath` option from the CLI, the server analyze
endpoint, and the programmatic entry through to `PipelineOptions`. Nothing
reads it yet; the reader added in the previous commit is still unwired.

Shaped deliberately after `springActuatorPath`, the existing option for an
out-of-band artifact: an explicit local path, accepting a directory or a
single file, resolved against the repository root so a committed
`docs/asyncapi` and an absolute cache populated by something else are both
natural, and `undefined` keeping the feature entirely off. Mirroring that
option rather than inventing a mechanism is what lets a downstream consumer
point the reader at documents fetched out of band without patching a file
here.

`analyze --watch` REJECTS the flag, exactly as it rejects --spring-actuator.
The watcher reacts to source changes and nothing watches a document
directory, so honouring it there would read the documents once and then
serve a stale answer for the rest of the session -- worse than refusing,
because it looks like it worked.

Additive only: 49 inserted lines, no deletions and no modified lines. Every
new interface member is optional and every forward is an object-literal
spread of an undefined value, so with the option unset the analyzer takes
byte-identical paths.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(ingestion): mint Destination nodes from AsyncAPI documents

Wires the reader into the destinations phase. With `--asyncapi-spec` set,
every `send` operation emits PUBLISHES_TO and every `receive` emits
CONSUMES_FROM against the ordinary resolved `Destination` node -- same key,
same `address` property -- so a document and a source site that name one
address on one broker land on ONE node and the two halves of a conversation
meet. Verified end to end: an address named only in a document is minted
with the right broker and direction, and an address a source site already
resolved stays a single node with its own `literal` provenance while the
same address on a different broker stays separate.

This claims only what a document states -- that the service talks to that
address, on that broker, in that direction -- and never which method does
it. The addresses in one document partition by (broker, action) into buckets
that usually hold more than one operation, so any assignment past a bucket
of size one is a heuristic, and a wrong one attaches a real address to the
wrong handler: a false connection wearing the clothes of a resolved one. The
edge therefore starts at the document, not at a callable. That is weaker
than a source-derived edge and worth having anyway, because it is available
where the source supplies nothing at all -- a programmatically registered
listener, a broker with no patterns here, a language whose messaging idiom
nobody has taught this codebase yet.

Documents are read even when the source pass found no messaging, which is
why the early return had to move: a repository whose brokers are invisible
to the patterns is precisely the case a published document covers, and an
early return keyed on source sites skipped the documents exactly there.

Their counters are kept in their own block rather than folded into the
existing ones. `refusalsByReason` is the denominator of the SOURCE
unresolved fraction, and a mistyped specification directory must not be able
to make the source look worse than it is. The block is absent -- not zeroed
-- when no path was configured, so "not asked for" stays distinguishable
from "asked for and found nothing"; those need different answers from an
operator and one zero cannot say which happened.

The direction assertion is the one that matters and it is pinned by type,
not by existence: inverting the mapping in the source tree fails exactly one
test, because every other assertion passes identically under both readings.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs: document --asyncapi-spec and why step 4 of the cascade stays empty

Adds the flag to both READMEs and to the three byte-identical copies of the
CLI skill, which a sync test pins together.

Also rewrites the note on the `specification` seam in the address cascade.
It said "nothing supplies it today", which was true and is now misleading:
a reader exists, and the hook is still unsupplied because of a decision
rather than for want of one.

A document names addresses; it does not name the method that uses one. To
hand an address to a particular candidate something must choose which of the
document's operations belongs to it, and the only division both sides agree
on -- (broker, action) -- leaves buckets that usually hold more than one
operation. On a real generated document exactly one bucket of four was
unambiguous. Every assignment past a bucket of size one is a heuristic, and
a wrong one puts a REAL address on a joining node under the wrong site: a
false connection wearing the clothes of a resolved one, which is the outcome
the keying rule exists to prevent.

The note also records the two things that would change that and are not
heuristics -- a document carrying the implementing symbol, or a
configuration source answering the `${key}` the candidate already recorded
-- and that the second wants its own resolver, since what it needs is the
placeholder key rather than the candidate.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(ingestion): refuse the document shapes that would forge a join

Four ways a conformant AsyncAPI document could mint a Destination that
connects two services which have said nothing about each other. Every one is
reachable from ordinary 3.x vocabulary, not from malformed input, and each is
now a countable refusal.

A PARAMETERIZED ADDRESS is a pattern, not a place. Two services that publish
`{env}.orders` share a template; one deploys with env=prod and the other with
env=staging, and keying on the template text merges them into a single node
with a publisher on one side and a subscriber on the other. This is the
document-side twin of `overridable-config-default`, which argues the same
thing about `${key:default}` in source. A channel declaring `parameters` is
the specification's own statement that its address is a template, so the
detector is a reading rather than a guess; the `{` test catches generators
that template without declaring.

ANY BINDINGS KEY WAS TAKEN AS A PROTOCOL. AsyncAPI allows `bindings` to be a
Reference Object, so the map's own key can be `$ref` -- and passed through,
that becomes half of a join key carrying no broker information at all. Two
services that both reference shared bindings and both name `orders` then land
on one node, defeating the broker-in-key rule that keeps `kafka orders` and
`rabbit orders` apart. A broker must now be spelled like a protocol name.

A BROKER CONTAINING A SPACE COLLIDES, because the node key joins with one:
("kafka orders", "x") and ("kafka", "orders x") are the same key. That was
latent while every broker came from Spring's closed union. This module is the
first caller to feed the shared helper text that a document wrote, which is
exactly the condition under which it stops being latent, so it is closed here
-- at the producer -- rather than by changing an encoding that `routeNodeKey`
shares.

THE ADDRESS WAS TRIMMED, while the source cascade keeps an address exactly as
written so `" orders "` stays its own node. Two producers of one key held
opposite whitespace policies and the document side erred toward joining.

Also: fold the transport-security protocol variants (`kafka-secure`,
`secure-mqtt`, `wss`, `stomps`, `https`) onto their base protocol. The
`amqp`->`rabbit` argument already in this file demands it -- AsyncAPI's server
vocabulary distinguishes them and its bindings vocabulary does not, so a
secured cluster's own document was being read as self-contradictory. Treat a
channel with no `servers` as available on all of the document's servers, which
is the specification's default and was costing every single-server document
its destinations. Bound the address and operation-id lengths, because
`generateId` concatenates rather than hashes, and bound total operations
across the run rather than only per document.

Read each file through ONE handle for both the size gate and the read, as
`actuator-runtime.ts` does and for the reason its comment gives (CodeQL
js/file-system-race): re-resolving the path lets a swapped file bypass the
cap, and the out-of-band cache this option reads is written by other tooling
by definition. Open it with O_NONBLOCK: the type check that rejects a FIFO is
unreachable without it, because opening a FIFO for reading blocks in open(2)
until a writer appears -- found by writing the test first and watching it time
out rather than fail.

Count symlinked entries and walk truncation instead of dropping them in
silence. A symlinked cache and a wrong path were producing identical results.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(analyze): rebuild when documents are configured, and report what was read

An AsyncAPI document is external to git freshness in exactly the way an
Actuator snapshot is: replacing one moves no commit and dirties no file. The
option's own README paragraph advertises an absolute cache written by other
tooling, and on the second run of that workflow the already-up-to-date fast
path fired, no document was ever opened, and the previous run's addresses were
served as current. Measured, not reasoned: editing a document and re-running
printed "Already up to date" and left the old address in the graph; with this
change the same probe re-reads and the new address replaces it.

So an enabled run forces a rebuild and dropping the option forces one more, to
clear document-derived evidence -- the treatment `springActuatorPath` already
gets, for the same reason. Only the FLAG is recorded in index metadata, not
the path: Actuator retains its inputs so future scans keep excluding them,
whereas a committed document is deliberately NOT excluded (it wants its real
`File` node), so there is nothing to retain and recording the path would put
an operator's directory layout into metadata for no consumer.

That also settles a defect it would have been tempting to patch separately. A
synthetic `File` node for an out-of-tree document carries a path that is in no
write set and is not covered by `isGraphWideNode`, so an incremental writeback
dropped the node while keeping its edges -- which then COPY against a row that
was never written, and fail into an IGNORE_ERRORS retry that reports success.
A forced rebuild has no incremental subgraph to get that wrong.

Distinguish the two meanings of `resolution: 'specification'`. That value
belongs to the address cascade and means a CODE candidate was resolved through
the step-4 hook; a node minted from a document has no code site and now says
`asyncapi-document`. Reusing one string would leave a query that groups by
provenance unable to separate an address a document states from one a document
was used to resolve, and only the second is a claim about source.

Report what was read. The stats block was justified on the grounds that an
operator must be able to tell a mistyped directory from a repository with no
documents -- and nothing surfaced it, so the justification was aspirational. A
configured path that yields nothing, or a walk that hit a bound, now warns
unconditionally, as `spring-auto-configuration.ts` does for the same class of
input. The phase summary carries the refusal breakdown rather than only the
totals, because the unresolved fraction is the number this work is judged on
and a bare count says how big the gap is without saying what would close it.

Tests for the three wiring lines that were individually deletable with a green
suite, following the templates already in the repository: a row in the
`--watch` rejection table, the CLI-threading assertion beside the Actuator
one, and the shipped-skill fragment that pins the flag in all three copies.
Also pin `filePath: ''` on a spec-minted destination -- the half of the keying
rule that stops a shared node becoming collateral damage of one document's
next change -- and the in-repo `File` branch, which was dead-code-able.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(ingestion): close the join-forging paths a second review round found

The `$ref` exclusion added last round fixed one instance of a class and left
the class open. A `bindings` map key of `x-scs-function` -- an ordinary
Specification Extension, which generators emit -- still became the broker, so
two unrelated services carrying one vendor annotation and one address landed
on ONE node whose broker half said nothing about any broker. And
`{ kafka: {}, x-internal: {} }` read as two brokers, losing a conformant
document and reporting it as self-contradictory, which also made any document
author a one-line saboteur of their own cross-service links.

So the two readers are now separate functions with opposite defaults, and the
header says why they must be. `servers[].protocol` is a FIELD DECLARED to hold
a protocol: an unrecognized value there is the document's own claim and passes
through, because refusing it would lose a destination the document states
plainly. A `bindings` MAP KEY is not that -- the specification puts `$ref` and
`x-` in the same namespace -- so a non-protocol key is the EXPECTED case and
only AsyncAPI's binding vocabulary may answer. The syntactic test that was
applied to both was the right rule for one of them.

The walk fix from last round introduced something worse than it reported. A
shared abort flag meant depth exhaustion in ONE branch terminated the whole
traversal, so ten good documents beside a twelve-deep unrelated subtree were
kept or lost depending on whether that subtree sorted before or after them.
Truncation and budget-exhaustion are now separate: depth returns from its own
branch, and only a genuinely global bound stops the walk.

Rewriting `protocol.ts` dropped the whitespace check from the server-protocol
path and made the node-key collision reachable again. The test written for
that collision last round caught it within the minute; the comment now records
that it was learned twice.

Everything else measured this round:

- The broker is the THIRD string that reaches a graph identifier, and it was
  unbounded while the header claimed there were two. A one-megabyte protocol in
  a document satisfying every other cap was measured producing a gigabyte of
  resident identifier strings, because `generateId` concatenates rather than
  hashes and the phase mints one id per node and per edge.
- The run-wide operation budget counted ACCEPTED operations, reproducing at the
  run level the exact defect the per-document cap was corrected for last round:
  a run whose every operation is refused never decrements it. Both now count
  operations EXAMINED, and `operation-cap` sets `truncated` -- it is a bound
  that stopped the operation count, which is what that flag is documented to
  mean.
- The channel-inherits-all-servers rule ran per operation. Hoisted: it depends
  only on the servers.
- A subdirectory that cannot be listed is counted rather than dropped, so a
  mixed-permission cache cannot report a clean, complete read.
- The read LOOPS, like the Actuator reader this claims to follow. A single read
  was never short across seven hundred probes on APFS, but POSIX permits it and
  FUSE mounts with `direct_io` -- the deployment this option targets -- return
  short counts. A document truncated at a line boundary still parses, so the
  failure is silent: operations vanish with `refusals: {}`.
- `parameters: {}` no longer refuses a literal address; an empty container
  states nothing and generators emit them.
- A channel that is itself a Reference Object gets its own reason instead of
  `no-address`, which was telling operators their documents omit addresses when
  the reader simply stops one hop short.
- A multi-protocol document resolves from its operation's own bindings; only
  when those are silent does an inherited multi-protocol server set refuse, and
  under `ambiguous-server-default` rather than a reason that says the document
  contradicts itself. It does not.
- HTTP and WebSocket are refused for destination minting. For a broker the
  topic is the namespace; for HTTP the host is, so keying on the path alone
  would make every service exposing `/events` one node. A `Route` already
  models an HTTP endpoint, with its method in the key.
- The sniff window is a parse gate, not a read gate, and four kilobytes refused
  a good document behind a licence header.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* test(ingestion): pin the wiring and the reporting that were deletable

Three lines could be deleted with the whole suite green, and each of them
makes the feature partly or wholly inert: the forward from `run-analyze` into
`PipelineOptions`, the forced rebuild while documents are configured, and the
cleanup rebuild when the option is dropped. None is visible one layer up,
where the CLI test asserts on a mock's arguments.

One integration test closes all three. It drives the real `runFullAnalysis`
against a real repository and asserts, in order: the enabled run does not take
the up-to-date fast path and logs the rebuild; the destination reaches the
graph, which only happens if the option is forwarded; a document edited with
the tree clean and the commit unchanged is re-read; dropping the option
rebuilds once and removes the document-derived evidence; and the run after
that is up to date again -- the "rebuilds once" half, which rests on the
metadata being written as a fresh literal rather than merged, and which
nothing pinned.

The document lives OUTSIDE the repository on purpose. That is the workflow the
option is documented for, and it is the only one where the hazard exists:
editing a tracked file dirties the tree and forces a rebuild anyway, so an
in-repo fixture would pass with the freshness fix reverted. Verified by
reverting both: deleting the forward fails on the empty destination list,
deleting the forced rebuild fails on the missing log line.

Also pinned, each because deleting the code it covers left the suite green:
the `parameters` half of the templated-address refusal (its old test supplied
a braced address too, so the `{` half alone satisfied it); the phase actually
forwarding `symlinksSkipped`; the unconditional warning, whose whole argument
is that a tally nobody can see is not a tally -- captured through the
repository's own `_captureLogger`; a `.yml` document; a character device,
which is the case the `isFile` check exists for and which the FIFO test does
not reach; and the bound that stops a walk.

Reject an empty `--asyncapi-spec` at the CLI. It resolved to the repository
root and walked the whole tree, defeating this module's own rule that there is
no glob-based auto-discovery -- and the HTTP entry point already rejected the
identical value. Two doors onto one option must not hold different rules.

Surface the flag in the MCP context resource beside `spring_actuator`. It
matters more there than for its neighbour: Actuator annotates nodes the source
pass already found, while document reading mints destinations and edges with
no code site, and nothing said where they came from.

Log the configured path relative to the repository. The same change refuses to
persist that path to index metadata because it would record an operator's
directory layout; holding that rule for metadata and not for logs was holding
it in one place.

Both test files now clean up their temporary directories.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* style(ingestion): apply the repository's prettier contract

`quality / format` runs `npx prettier --check .`, and three files added by this
branch were not formatted to it. No behaviour changes: the reader's line
breaks and two test literals move, nothing else.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* test(ingestion): release the mini-repo handle the document test allocated

`setupMiniRepo` documents that the caller owns cleanup, and every other test
in this file calls `repo.cleanup()` in its `finally`. The AsyncAPI document
test removed only the document directory, so each run left a temporary
repository behind.

The two owners are separate on purpose: the document directory is a SIBLING of
the repository, placed outside the working tree so that editing it cannot
dirty the tree and force a rebuild on its own. The repo's cleanup therefore
does not reach it, and both calls belong in the same block.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(ingestion): stop partial server evidence from reading as unanimous

Six review findings, every one a way this reader could name a broker the
document does not name. They share a shape: something is DROPPED rather than
refused, the remaining evidence agrees with itself, and an operation is
attributed with confidence to a broker its document never settled on. A wrong
broker is half a join key, so it does not produce a missing edge -- it produces
an edge to a stranger, reported as a fact.

CAPPED SERVER MAPS. A channel with no `servers` inherits all of them, and that
map is capped at 1,000. A document whose first thousand servers are Kafka and
whose thousand-and-first is JMS read as unanimously Kafka, because unanimity
was tested on the slice. Counted `server-cap` and set `truncated`, but neither
stopped the attribution. The inherited path now refuses under
`capped-server-default` -- checked BEFORE agreement, since a subset agrees with
itself for free.

ROOT SERVER REFERENCE OBJECTS. The Servers Object patterned field is
`Server Object | Reference Object`, so `{ $ref: '#/components/servers/prod' }`
is conformant. Reading `protocol` off the raw value dropped every one: an
all-reference document had no protocol at all, and -- worse -- a MIXED set lost
its disagreeing half and became unanimous. One hop is now followed, through
`#/servers` and `#/components/servers`; anything else is refused under
`unresolved-server-reference` rather than skipped.

CHANNEL BINDINGS. Only the operation's bindings were read. A conformant channel
carrying `bindings: { kafka: {} }` with no operation binding was dropped as
`protocol-unknown` while the document said plainly which broker it meant, and a
disagreement between the two levels was invisible. Both are read; a conflict is
`protocol-disagreement`.

EMPTY `servers`. "If `servers` is absent or empty, this channel MUST be
available on all the servers defined in the Servers Object" -- one sentence,
both cases. A zero-iteration loop returned `explicit: true`, which blocked the
inherited fallback and dropped valid operations.

POINTER DECODING ORDER. RFC 6901 percent-decodes the fragment BEFORE splitting
on `/`. The raw token was tested for a separator first, so `#/channels/orders%2Fv1`
passed a check it should have failed and then decoded into two segments -- a
pointer addressing `channels.orders.v1` was read as a channel named `orders/v1`,
inventing a channel the document never declared. A malformed escape is now
refused rather than resolved against its undecoded text. `~1` still resolves; it
is the pointer's own escape and belongs after segmentation.

THE SNIFF WINDOW. A fixed window decides by where the root key sits rather than
whether it is there, so every window is a false negative waiting for a longer
preamble -- 4 KiB was replaced by 64 KiB for that reason and inherited the same
defect. The whole text is scanned; it is already bounded and already in memory,
and the gate exists to skip the PARSE, which is the expensive half. A leading
UTF-8 BOM is stripped before both sniff and parse.

Twelve of the fourteen new tests were run against the unfixed reader and all
twelve failed; the other two are controls that must pass either way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* refactor(ingestion): simplify AsyncAPI pointer and binding resolution

Decode each $ref once, union binding evidence, and stop walking capped
server maps whose brokers are unused on the inherit path.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Gergő Magyar <gergomagyar@icloud.com>
Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

.claude/skills/gitnexus-cli/SKILL.md

@@ -28,6 +28,7 @@ Run from the project root. This parses all source files, builds the knowledge gr
 | `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
 | `--pdg` | Build the program-dependence layers used by `explain` and `pdg_query` (taint, CDG, and REACHING_DEF). |
 | `--spring-actuator <path>` | Import opt-in Spring Boot Actuator mappings, beans, conditions, configprops, and env snapshots. Forces a full rebuild; unsupported with `--watch`. |
+| `--asyncapi-spec <path>` | Read opt-in AsyncAPI 3.x documents (directory or single file) and mint `Destination` nodes from their operations. 2.x is refused, not mapped. Unsupported with `--watch`. |
 
 **When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
 

README.md

@@ -450,12 +450,19 @@ gitnexus analyze --worker-timeout 60  # Increase worker idle timeout for slow pa
 gitnexus analyze --workers <n>   # Parse worker pool size (>=1; default: cores-1, capped at 16,
                                  # auto-sized to the repo). 0 is rejected — there is no sequential mode.
 gitnexus analyze --spring-actuator ./actuator  # Enrich with local Spring Boot Actuator JSON snapshots
+gitnexus analyze --asyncapi-spec ./docs/asyncapi  # Resolve broker addresses from AsyncAPI 3.x documents
 gitnexus analyze --wal-checkpoint-threshold 67108864  # LadybugDB WAL auto-checkpoint threshold in bytes
                                  # (default 67108864 = 64 MiB; -1 keeps Ladybug stock ~16 MiB)
 ```
 
 `--spring-actuator` is explicitly opt-in and accepts either a JSON bundle keyed by `mappings`, `beans`, `conditions`, `configprops`, and/or `env`, or a directory containing endpoint-named JSON files. It confirms matching static nodes and adds conservative runtime-only routes, beans, and property keys. The configured input is excluded from source scanning; only normalized repository-relative exclusions are retained for future scans, never absolute paths. Env/configprops values, origins, condition messages, and source names are never persisted or printed. Because snapshots are external runtime state, an enabled run always rebuilds; the first later run without the option rebuilds once to remove runtime evidence. The same path can be set as `springActuator` in `.gitnexusrc`.
 
+`--asyncapi-spec` is explicitly opt-in and accepts a directory of AsyncAPI documents or a single document; the path is resolved against the repository root, so a committed `docs/asyncapi` and an absolute cache written by something else both work. Each `operations[]` entry of an **AsyncAPI 3.x** document can contribute a `Destination` node keyed by broker and address, with `action: send` emitting `PUBLISHES_TO` and `action: receive` emitting `CONSUMES_FROM`, so a document and source code that name one address on one broker land on the same node. Edges start at the document, not at a callable — a document states that the service talks to an address, not which method does — and no address a document names is ever attached to an unresolved source site.
+
+An operation must name a protocol, either through its own `bindings` or through the `servers[].protocol` of the servers its channel resolves to (a channel that lists no `servers` resolves to all of them); operations that name none are refused, as are operations whose two readings name different brokers, and channels that inherit a multi-protocol server set without choosing. HTTP and WebSocket documents are refused for destination minting: there the host rather than the address names the place, and an HTTP endpoint is already modelled as a `Route`. A parameterized address — a channel declaring `parameters`, or an address containing `{` — is refused rather than keyed: two services publishing `{env}.orders` share a pattern, not a queue. AsyncAPI **2.x is refused** under its own counted reason and never mapped, because its `publish`/`subscribe` are inverted relative to 3.x `send`/`receive` and a naive mapping would reverse the async graph while leaving it connected. Every refusal is counted, and a configured path that yields nothing is reported rather than passed over in silence.
+
+Like Actuator snapshots, documents are external to git freshness — replacing one moves no commit and dirties no file — so an enabled run always rebuilds, and the first later run without the option rebuilds once to remove document-derived evidence. There is no glob-based auto-discovery, and the option is unsupported with `--watch`.
+
 If `analyze` reports a worker parse timeout on a large or unusual repository, it keeps running and falls back safely. To give slow worker jobs more time, use `--worker-timeout 60` or set `GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=60000`. For very large files, `GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES` controls the worker job byte budget.
 
 **Embeddings node limit** — `gitnexus analyze --embeddings` generates semantic search vectors with a default 50,000-node safety cap to protect memory on large repositories:

gitnexus-claude-plugin/skills/gitnexus-cli/SKILL.md

@@ -28,6 +28,7 @@ Run from the project root. This parses all source files, builds the knowledge gr
 | `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
 | `--pdg` | Build the program-dependence layers used by `explain` and `pdg_query` (taint, CDG, and REACHING_DEF). |
 | `--spring-actuator <path>` | Import opt-in Spring Boot Actuator mappings, beans, conditions, configprops, and env snapshots. Forces a full rebuild; unsupported with `--watch`. |
+| `--asyncapi-spec <path>` | Read opt-in AsyncAPI 3.x documents (directory or single file) and mint `Destination` nodes from their operations. 2.x is refused, not mapped. Unsupported with `--watch`. |
 
 **When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
 

gitnexus/README.md

@@ -349,6 +349,12 @@ anchors are deliberately omitted. Add common infrastructure fields such as `/hea
 
 `--spring-actuator` is explicitly opt-in. The path may be a JSON bundle keyed by `mappings`, `beans`, `conditions`, `configprops`, and/or `env`, or a directory containing endpoint-named JSON files. Runtime mappings and beans confirm matching static nodes; conditions and configuration property keys enrich existing evidence, with conservative runtime-only nodes added when no match exists. The configured input is excluded from source scanning; only normalized repository-relative exclusions are retained for future scans, never absolute paths. Env/configprops values, origins, condition messages, and source names are never persisted or printed. Enabled runs always rebuild because runtime snapshots are external to git freshness; omitting the option later rebuilds once to remove runtime evidence. Project config can set the same path with `springActuator` in `.gitnexusrc`.
 
+`--asyncapi-spec` is explicitly opt-in and accepts a directory of AsyncAPI documents or a single document; the path is resolved against the repository root, so a committed `docs/asyncapi` and an absolute cache written by something else both work. Each `operations[]` entry of an **AsyncAPI 3.x** document can contribute a `Destination` node keyed by broker and address, with `action: send` emitting `PUBLISHES_TO` and `action: receive` emitting `CONSUMES_FROM`, so a document and source code that name one address on one broker land on the same node. Edges start at the document, not at a callable — a document states that the service talks to an address, not which method does — and no address a document names is ever attached to an unresolved source site.
+
+An operation must name a protocol, either through its own `bindings` or through the `servers[].protocol` of the servers its channel resolves to (a channel that lists no `servers` resolves to all of them); operations that name none are refused, as are operations whose two readings name different brokers, and channels that inherit a multi-protocol server set without choosing. HTTP and WebSocket documents are refused for destination minting: there the host rather than the address names the place, and an HTTP endpoint is already modelled as a `Route`. A parameterized address — a channel declaring `parameters`, or an address containing `{` — is refused rather than keyed: two services publishing `{env}.orders` share a pattern, not a queue. AsyncAPI **2.x is refused** under its own counted reason and never mapped, because its `publish`/`subscribe` are inverted relative to 3.x `send`/`receive` and a naive mapping would reverse the async graph while leaving it connected. Every refusal is counted, and a configured path that yields nothing is reported rather than passed over in silence.
+
+Like Actuator snapshots, documents are external to git freshness — replacing one moves no commit and dirties no file — so an enabled run always rebuilds, and the first later run without the option rebuilds once to remove document-derived evidence. There is no glob-based auto-discovery, and the option is unsupported with `--watch`.
+
 > **`gitnexus uninstall`** reverses `gitnexus setup` — it removes the GitNexus MCP entries, hooks, and skill directories it added to each detected editor. Skill directories are identified **by bundled gitnexus skill name** (e.g. `gitnexus-cli/`), so if you customized files inside an installed skill directory, back them up first. It is a dry-run preview by default and prints the exact paths it would remove; pass `--force` to apply. Per-repo indexes (`gitnexus clean --all`) and the global npm package (`npm uninstall -g gitnexus`) are left for you to remove.
 
 ## Remote Embeddings

gitnexus/skills/gitnexus-cli.md

@@ -28,6 +28,7 @@ Run from the project root. This parses all source files, builds the knowledge gr
 | `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
 | `--pdg` | Build the program-dependence layers used by `explain` and `pdg_query` (taint, CDG, and REACHING_DEF). |
 | `--spring-actuator <path>` | Import opt-in Spring Boot Actuator mappings, beans, conditions, configprops, and env snapshots. Forces a full rebuild; unsupported with `--watch`. |
+| `--asyncapi-spec <path>` | Read opt-in AsyncAPI 3.x documents (directory or single file) and mint `Destination` nodes from their operations. 2.x is refused, not mapped. Unsupported with `--watch`. |
 
 **When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
 

gitnexus/src/cli/analyze-options.ts

@@ -129,6 +129,13 @@ export interface AnalyzeOptions {
    * bundle or a directory containing endpoint JSON files. Disabled by default.
    */
   springActuator?: string;
+  /**
+   * Explicit local AsyncAPI 3.x document input. Accepts a directory of
+   * documents or a single document, resolved against the repository root so an
+   * out-of-band cache and a committed directory are equally usable. Disabled by
+   * default.
+   */
+  asyncapiSpec?: string;
   /** OpenAI-compatible embeddings base URL (incl. /v1). Overrides GITNEXUS_EMBEDDING_URL. */
   embeddingBaseUrl?: string;
   /** Embedding model name. Overrides GITNEXUS_EMBEDDING_MODEL. */

gitnexus/src/cli/analyze-watch.ts

@@ -116,6 +116,11 @@ export async function resolveWatchOptions(
     ['--index-only', cli.indexOnly],
     ['--skip-git', cli.skipGit],
     ['--spring-actuator', cli.springActuator],
+    // Rejected under --watch for the same reason as --spring-actuator: the
+    // watcher reacts to source changes, and nothing watches an out-of-band
+    // document directory. Honouring the flag here would read the documents once
+    // and then quietly serve a stale answer for the rest of the session.
+    ['--asyncapi-spec', cli.asyncapiSpec],
     ['walCheckpointThreshold', cli.walCheckpointThreshold],
     ['embeddingThreads', cli.embeddingThreads],
     ['embeddingBatchSize', cli.embeddingBatchSize],

gitnexus/src/cli/analyze.ts

@@ -1006,6 +1006,17 @@ const analyzeCommandImpl = async (
     return;
   }
 
+  // An empty value resolves to the repository root, so `--asyncapi-spec ""`
+  // walks the whole tree — defeating the module's own rule that there is no
+  // glob-based auto-discovery, and spending the walk budget on `node_modules`.
+  // The HTTP entry point already rejects exactly this value; two doors onto one
+  // option must not hold different rules.
+  if (options.asyncapiSpec !== undefined && options.asyncapiSpec.trim() === '') {
+    cliError('  --asyncapi-spec must be a non-empty path.\n');
+    process.exitCode = 1;
+    return;
+  }
+
   if (options.embeddingDevice) {
     const allowed = new Set(['auto', 'cpu', 'dml', 'cuda', 'wasm']);
     if (!allowed.has(options.embeddingDevice)) {
@@ -1375,6 +1386,7 @@ const analyzeCommandImpl = async (
       // forwarded to the routes phase consumer scan.
       fetchWrappers: options.fetchWrappers,
       springActuatorPath: options.springActuator,
+      asyncApiSpecPath: options.asyncapiSpec,
       // The CLI always process.exit()s after this returns (success path at the
       // end of analyzeCommandImpl, error/interrupt paths via process.exit too),
       // so the finalize close skips the native conn/db close — it can double-free

gitnexus/src/cli/index.ts

@@ -163,6 +163,11 @@ program
     'Import local Spring Boot Actuator JSON snapshots (mappings, beans, conditions, ' +
       'configprops, env). Explicit opt-in; disabled by default.',
   )
+  .option(
+    '--asyncapi-spec <path>',
+    'Read AsyncAPI 3.x documents from this directory or file and resolve broker ' +
+      'addresses from them. Explicit opt-in; disabled by default.',
+  )
   .option('--embedding-threads <n>', 'Limit local ONNX embedding CPU threads')
   .option('--embedding-batch-size <n>', 'Number of nodes per embedding batch')
   .option('--embedding-sub-batch-size <n>', 'Number of chunks per embedding model call')

gitnexus/src/core/ingestion/asyncapi/document.ts

@@ -0,0 +1,950 @@
+/**
+ * Read AsyncAPI 3.x documents off disk and normalize their operations into
+ * broker addresses.
+ *
+ * Deliberately OUTSIDE `frameworks/spring/`. An AsyncAPI document is a
+ * published artifact, not a Spring one: it is emitted by generators across
+ * Java, Kotlin, TypeScript, Go and Python toolchains, and it is written by hand
+ * as often as it is generated. The entry criterion here is therefore the
+ * DOCUMENT FORMAT — a root `asyncapi` key — and never the generator. Nothing in
+ * this module may branch on `x-generator`, on a vendor extension, or on the
+ * shape of an operation key: the moment it does, every service whose toolchain
+ * spells things differently stops being read, and the failure is silent.
+ *
+ * ── WHY THIS IS WORTH READING AT ALL ──────────────────────────────────────
+ *
+ * A `@KafkaListener(topics = "${app.topic.in}")` names a configuration key, not
+ * an address, and the address cascade correctly refuses to resolve it — two
+ * services that merely wrote the same placeholder have said nothing about each
+ * other. But the service's own published document states the address outright,
+ * fully resolved, because the generator ran with the configuration applied.
+ * That is a fact about the service that no amount of reading its source can
+ * recover.
+ *
+ * ── WHAT THIS MODULE IS AFRAID OF ─────────────────────────────────────────
+ *
+ * Everything it emits becomes half of a JOIN KEY. A destination minted here
+ * meets every other site in every other repository that names the same address
+ * on the same broker — that is the whole value, and it is the whole hazard. A
+ * missing destination is a visible gap; a wrong one is reported as a fact. So
+ * the refusals below are not defensive clutter: each one is a case where the
+ * document says something that LOOKS like an address and is not one, and where
+ * accepting it would connect two services that have said nothing about each
+ * other. The taxonomy is closed and countable for the same reason the source
+ * cascade's is — a feature judged on its unresolved fraction needs the fraction
+ * broken down by cause, or nobody can tell it what to go and fix.
+ *
+ * ── VERSION 2.x IS REFUSED, NOT MAPPED ────────────────────────────────────
+ *
+ * AsyncAPI 2.x describes a channel from the READER's point of view: `publish`
+ * means "you may publish here", so the documenting application RECEIVES, and
+ * `subscribe` means the application SENDS. Version 3.0 renamed these to the
+ * application's own `receive` / `send`. Mapping 2.x naively therefore reverses
+ * every direction in the async graph — and reverses it INVISIBLY, because both
+ * roles still exist, every edge is still emitted, and the graph stays
+ * connected. Nothing fails; the arrows simply point the wrong way.
+ *
+ * The inversion is one line to write and impossible to test against a real
+ * corpus we do not have, and the 2.x wording confused implementers badly enough
+ * that some generators emitted it backwards. So 2.x is refused under its own
+ * countable reason instead. A silent skip would be indistinguishable from "this
+ * service publishes no document", which is the one thing the count has to be
+ * able to tell us: if the refusal tally shows 2.x documents in the field, the
+ * inversion earns its way in with evidence behind it.
+ */
+
+import { createRequire } from 'node:module';
+import fs from 'node:fs/promises';
+import { constants as fsConstants } from 'node:fs';
+import path from 'node:path';
+import { brokerForBindingKey, brokerForProtocol, isNonDestinationBroker } from './protocol.js';
+
+// `js-yaml` is CJS; the rest of this repository reaches it the same way
+// (`pipeline-phases/spring-config.ts`, `import-resolvers/node-workspace-packages.ts`).
+const _require = createRequire(import.meta.url);
+const yaml = _require('js-yaml') as typeof import('js-yaml');
+
+/**
+ * A published document is data, not code, so it is parsed under the JSON
+ * schema — the same choice `core/group/config-parser.ts` makes for `group.yaml`.
+ * No custom tags, no timestamps, no `yes`/`no` booleans: an address is whatever
+ * the document literally spells, and nothing may be coerced into another type
+ * on the way in.
+ */
+const DOCUMENT_SCHEMA = yaml.JSON_SCHEMA;
+
+/** Generous for a specification, small enough that a mistake is caught. */
+const MAX_DOCUMENT_BYTES = 8 * 1024 * 1024;
+/** Bounded so one pathological document cannot dominate a run. Counted against
+ *  operations EXAMINED, not accepted: a document with a hundred thousand
+ *  refused operations costs the same walk as one with a hundred thousand good
+ *  ones, and a cap that only counts successes does not bound the work. */
+const MAX_OPERATIONS_PER_DOCUMENT = 5_000;
+/** The same bound across the whole run, and counted the same way — EXAMINED,
+ *  not accepted. Counting successes here reproduced the very defect the
+ *  per-document cap was corrected for: a run whose every operation was refused
+ *  never decremented the budget, so all two thousand documents were processed
+ *  in full and the result still reported `truncated: false`. */
+const MAX_TOTAL_OPERATIONS = 50_000;
+/** Bounded so a mis-aimed path (a whole repository, `/`) cannot walk forever. */
+const MAX_DOCUMENTS = 2_000;
+/** Directory entries VISITED, not documents accepted. The document cap alone
+ *  bounds nothing on a tree that contains no documents. */
+const MAX_WALK_ENTRIES = 100_000;
+const MAX_DIRECTORY_DEPTH = 8;
+/** Servers per document. The channel-inherits-all-servers rule reads this map,
+ *  and YAML aliases make a server about sixteen bytes, so an in-cap document
+ *  can declare hundreds of thousands of them. */
+const MAX_SERVERS_PER_DOCUMENT = 1_000;
+/**
+ * A leading byte-order mark, stripped before the file is sniffed or parsed.
+ *
+ * An editor that saves UTF-8 with a BOM puts one code point in front of the
+ * root key, which is enough to make the sniff miss and refuse a perfectly good
+ * document as `not-a-document`.
+ */
+const BOM = '\uFEFF';
+/**
+ * An address and an operation id both end up inside graph identifiers, and
+ * `generateId` CONCATENATES rather than hashes (`lib/utils.ts`), so an
+ * identifier is exactly as long as the text it was built from. One document
+ * under every other cap — a multi-megabyte address plus five thousand
+ * operations naming it — therefore mints five thousand multi-megabyte edge ids,
+ * each flattened into a string key by the graph's `Map`.
+ *
+ * The BROKER is the third such string and is bounded in `protocol.ts`; the
+ * count matters because an earlier version of this comment said "the two
+ * strings that reach an id", left the third unbounded, and a one-megabyte
+ * protocol was measured turning a one-megabyte document into a gigabyte of
+ * resident identifiers.
+ */
+const MAX_ADDRESS_LENGTH = 2_048;
+const MAX_OPERATION_ID_LENGTH = 512;
+
+const DOCUMENT_EXTENSIONS: ReadonlySet<string> = new Set(['.yaml', '.yml', '.json']);
+
+/**
+ * Why a document, or one operation inside it, produced no address.
+ *
+ * A CLOSED, COUNTABLE set, and deliberately NOT `SpringDestinationRefusal`.
+ * That union is documented as the reasons a *source-level candidate* produced
+ * no address, and it is the denominator of the unresolved fraction the address
+ * work is judged on. Folding document-level failures into it would silently
+ * change what that number means — a repository whose specification directory
+ * was mistyped would report a worse SOURCE, which is the opposite of the truth.
+ *
+ * Members are split wherever two causes are different FACTS about the input.
+ * A tally whose member says "the document contradicts itself" when the document
+ * is merely multi-protocol sends an operator to fix the wrong thing, and this
+ * tally is the number the whole feature is judged on.
+ */
+export type AsyncApiRefusal =
+  /** The file parsed but has no root `asyncapi` key: not a document at all. */
+  | 'not-a-document'
+  /** Root `asyncapi: 2.x`. See the header — refused, never mapped. */
+  | 'asyncapi-2-unsupported'
+  /** A root `asyncapi` key naming a version this module does not read. */
+  | 'unsupported-version'
+  /** Malformed YAML/JSON, or a root that is not an object. */
+  | 'unparsable'
+  /** The file could not be read, or is not a regular file (a FIFO, a device). */
+  | 'unreadable'
+  /** A subdirectory could not be listed. Counted rather than skipped: under a
+   *  mixed-permission cache half the documents can be invisible while the run
+   *  otherwise reports a clean, complete read. */
+  | 'directory-unreadable'
+  /** Larger than {@link MAX_DOCUMENT_BYTES}. */
+  | 'oversized'
+  /** The document held more operations than one run will examine. */
+  | 'operation-cap'
+  /** The run as a whole reached {@link MAX_TOTAL_OPERATIONS}. */
+  | 'total-operation-cap'
+  /** The document declares more servers than the channel-inheritance rule will
+   *  read. */
+  | 'server-cap'
+  /** The walk hit a bound before it finished, so the document set is a floor
+   *  rather than the whole of what the configured path holds. A truncated read
+   *  that reported nothing would be indistinguishable from a complete one. */
+  | 'walk-truncated'
+  /** `operations[].channel.$ref` is absent or not a local channel pointer. */
+  | 'no-channel-reference'
+  /** The `$ref` resolved to no channel in this document. */
+  | 'channel-not-found'
+  /** The channel entry is itself a Reference Object, which this module does not
+   *  follow. Distinct from `no-address` on purpose: a `$ref`-ed channel HAS an
+   *  address, somewhere this reader did not look, and filing it under
+   *  `no-address` tells an operator their documents omit addresses when the
+   *  real answer is that the reader stops one hop short. */
+  | 'unresolved-channel-reference'
+  /** The channel names no `address`, so there is nothing to key on. */
+  | 'no-address'
+  /**
+   * The address is a TEMPLATE, not an address: the channel declares non-empty
+   * `parameters`, or the address carries a `{…}` placeholder.
+   *
+   * This is the document-side twin of the source cascade's
+   * `overridable-config-default`, and it exists for the identical reason. Two
+   * services that both publish `{env}.orders` have named a pattern they share,
+   * not a queue they share — one deploys with `env=prod` and the other with
+   * `env=staging`, and keying on the template text merges them into a single
+   * node with a publisher on one side and a subscriber on the other. That is a
+   * false connection built entirely from conformant AsyncAPI: `parameters` and
+   * `{param}` are core 3.x vocabulary, not a vendor quirk.
+   */
+  | 'templated-address'
+  /** Longer than {@link MAX_ADDRESS_LENGTH}. */
+  | 'address-too-long'
+  /** Longer than {@link MAX_OPERATION_ID_LENGTH}. */
+  | 'operation-id-too-long'
+  /** `action` is neither `send` nor `receive`. */
+  | 'unrecognized-action'
+  /** Neither the operation's bindings nor the servers its channel resolves to
+   *  name a protocol. Silence about the broker is not a claim about it, but a
+   *  `Destination` cannot be keyed without one. */
+  | 'protocol-unknown'
+  /** The operation's OWN two statements about its broker — its bindings and the
+   *  servers its channel explicitly lists — name different brokers. The
+   *  document contradicts itself, and a destination keyed on the wrong broker
+   *  joins a stranger. */
+  | 'protocol-disagreement'
+  /** The channel lists no `servers`, so it inherits all of them, and they do
+   *  not agree on one broker. The document does NOT contradict itself here —
+   *  it is simply multi-protocol and this channel did not choose — which is why
+   *  this is not `protocol-disagreement`. */
+  | 'ambiguous-server-default'
+  /**
+   * The channel inherits the document's servers, but that map was CAPPED at
+   * {@link MAX_SERVERS_PER_DOCUMENT}, so the brokers read are a subset.
+   *
+   * Distinct from `ambiguous-server-default`, and the distinction is the whole
+   * point: unanimity across a subset is not unanimity. A document whose first
+   * thousand servers are Kafka and whose thousand-and-first is JMS reads as
+   * unanimously Kafka, and every operation inheriting it would be attributed to
+   * a broker the complete set does not agree on.
+   */
+  | 'capped-server-default'
+  /**
+   * A server this operation depends on is a Reference Object this reader could
+   * not resolve — a pointer outside `#/servers` and `#/components/servers`, a
+   * name that is absent, or a reference to another reference.
+   *
+   * Refused rather than skipped. Skipping one server of several silently
+   * narrows the evidence, and a narrowed set is what makes a mixed document
+   * look like it agrees with itself.
+   */
+  | 'unresolved-server-reference'
+  /** The broker is HTTP or WebSocket, where the host rather than the address is
+   *  the namespace. See `isNonDestinationBroker`. */
+  | 'not-a-destination-protocol';
+
+export interface AsyncApiOperation {
+  /** Absolute path of the document this operation came from. */
+  readonly documentPath: string;
+  /** The `operations` map key, kept for provenance and carried into the edge
+   *  `reason` so a reader can find the operation the edge came from. */
+  readonly operationId: string;
+  readonly action: 'send' | 'receive';
+  readonly address: string;
+  /** Normalized broker — the first half of the `Destination` key. */
+  readonly broker: string;
+}
+
+export interface AsyncApiReadResult {
+  readonly operations: readonly AsyncApiOperation[];
+  /** Files considered — every candidate extension under the configured path. */
+  readonly documentsScanned: number;
+  /** Files that parsed as an AsyncAPI 3.x document and yielded an operation. */
+  readonly documentsAccepted: number;
+  /** Entries skipped because they were symbolic links. Not a refusal — the skip
+   *  is deliberate — but counted, because a cache written by other tooling is
+   *  very often a symlink farm, and an operator whose whole cache was skipped
+   *  would otherwise see a result identical to a wrong path. */
+  readonly symlinksSkipped: number;
+  /** True when a bound stopped the walk or the operation count, so every number
+   *  here is a floor rather than a total. */
+  readonly truncated: boolean;
+  /** Every refusal, document-level and operation-level, by reason. */
+  readonly refusals: Readonly<Partial<Record<AsyncApiRefusal, number>>>;
+}
+
+interface Tally {
+  count(reason: AsyncApiRefusal): void;
+}
+
+function makeTally(sink: Partial<Record<AsyncApiRefusal, number>>): Tally {
+  return {
+    count: (reason) => {
+      sink[reason] = (sink[reason] ?? 0) + 1;
+    },
+  };
+}
+
+/**
+ * Own-property read that cannot be answered by the prototype chain.
+ *
+ * A document is untrusted input and its keys are attacker-chosen in the general
+ * case. `channels['constructor']` misses because {@link asRecord} rejects a
+ * function — but `channels['__proto__']` would otherwise resolve to
+ * `Object.prototype`, which IS an object and would sail through as an empty
+ * channel. This guard, not the type test, is what stops that one.
+ */
+function own(container: unknown, key: string): unknown {
+  if (typeof container !== 'object' || container === null) return undefined;
+  if (!Object.prototype.hasOwnProperty.call(container, key)) return undefined;
+  return (container as Record<string, unknown>)[key];
+}
+
+function asRecord(value: unknown): Record<string, unknown> | undefined {
+  if (typeof value !== 'object' || value === null || Array.isArray(value)) return undefined;
+  return value as Record<string, unknown>;
+}
+
+function asString(value: unknown): string | undefined {
+  return typeof value === 'string' ? value : undefined;
+}
+
+/** URI-fragment percent-decode. Malformed `%` sequences refuse the pointer. */
+function decodeFragment(ref: string): string | undefined {
+  try {
+    return decodeURIComponent(ref);
+  } catch (err) {
+    // Malformed percent-escapes throw URIError. Anything else is a real bug.
+    if (!(err instanceof URIError)) throw err;
+    return undefined;
+  }
+}
+
+/** RFC 6901's own escapes, `~1` before `~0` — a literal `~1` produced by
+ *  decoding `~01` would otherwise be mistaken for a slash. */
+function unescapePointerToken(token: string): string {
+  return token.split('~1').join('/').split('~0').join('~');
+}
+
+/**
+ * Trailing name of `<prefix><token>` on an already-decoded fragment.
+ *
+ * DECODE, THEN SEGMENT. RFC 6901 percent-decodes the URI fragment first; only
+ * then is the result split on `/`. Testing the raw text for a separator lets
+ * `#/channels/orders%2Fv1` through as one segment and then decode it into two,
+ * inventing a channel named `orders/v1`. A real slash in a name is `~1`.
+ */
+function nameAfterPrefix(decoded: string, prefix: string): string | undefined {
+  if (!decoded.startsWith(prefix)) return undefined;
+  const token = decoded.slice(prefix.length);
+  if (token === '' || token.includes('/')) return undefined;
+  return unescapePointerToken(token);
+}
+
+function pointerName(ref: string, prefix: string): string | undefined {
+  const decoded = decodeFragment(ref);
+  if (decoded === undefined) return undefined;
+  return nameAfterPrefix(decoded, prefix);
+}
+
+/**
+ * Distinct brokers named by a bindings object's own keys.
+ *
+ * Routed through `brokerForBindingKey`, which answers only for AsyncAPI's
+ * binding vocabulary — `$ref` and `x-` extensions share this namespace
+ * legitimately and are not brokers. See `protocol.ts` for why this differs from
+ * the pass-through applied to `servers[].protocol`.
+ */
+function brokersFromBindings(bindings: unknown): Set<string> {
+  const out = new Set<string>();
+  const record = asRecord(bindings);
+  if (record === undefined) return out;
+  for (const key of Object.keys(record)) {
+    const broker = brokerForBindingKey(key);
+    if (broker !== undefined) out.add(broker);
+  }
+  return out;
+}
+
+/**
+ * The broker one Servers Object entry names, following at most one local `$ref`.
+ *
+ * The Servers Object's patterned field is `Server Object | Reference Object`,
+ * so an entry may legitimately be `{ $ref: '#/components/servers/prod' }`.
+ * Reading `protocol` off the raw value drops every one of those, and a dropped
+ * server is not neutral here: in a mixed set it removes the disagreeing half
+ * and makes partial evidence look unanimous, which is exactly how a confident
+ * WRONG broker gets attributed.
+ *
+ * `unresolved` is reported rather than swallowed so the caller can refuse the
+ * attribution instead of answering from the servers it happened to understand.
+ * A reference to a reference counts as unresolved too: one hop covers every
+ * document shape seen in practice, and chasing a chain over untrusted input
+ * would need a cycle guard before it were safe at all.
+ */
+function serverBroker(
+  entry: unknown,
+  root: Record<string, unknown>,
+): { broker: string | undefined; unresolved: boolean } {
+  const ref = asString(own(entry, '$ref'));
+  if (ref === undefined) {
+    return { broker: brokerForProtocol(asString(own(entry, 'protocol'))), unresolved: false };
+  }
+  const target = resolveLocalServerRef(ref, root);
+  if (target === undefined || own(target, '$ref') !== undefined) {
+    return { broker: undefined, unresolved: true };
+  }
+  return { broker: brokerForProtocol(asString(own(target, 'protocol'))), unresolved: false };
+}
+
+/** `#/servers/<name>` or `#/components/servers/<name>` → that Server Object. */
+function resolveLocalServerRef(
+  ref: string,
+  root: Record<string, unknown>,
+): Record<string, unknown> | undefined {
+  const decoded = decodeFragment(ref);
+  if (decoded === undefined) return undefined;
+  const direct = nameAfterPrefix(decoded, '#/servers/');
+  if (direct !== undefined) return asRecord(own(asRecord(own(root, 'servers')), direct));
+  const inComponents = nameAfterPrefix(decoded, '#/components/servers/');
+  if (inComponents === undefined) return undefined;
+  const components = asRecord(own(asRecord(own(root, 'components')), 'servers'));
+  return asRecord(own(components, inComponents));
+}
+
+/**
+ * Brokers of the servers a channel names explicitly.
+ *
+ * An EMPTY array is not an explicit choice. The specification defines the two
+ * cases identically — "If `servers` is absent or empty, this channel MUST be
+ * available on all the servers defined in the Servers Object" — so reporting
+ * `explicit: true` after a zero-iteration loop blocks the inherited fallback
+ * and drops a perfectly valid operation as `protocol-unknown`.
+ *
+ * A channel's `servers` MUST hold Reference Objects — the specification says so
+ * in as many words, and forbids Server Objects there by name — so an entry that
+ * is not a resolvable local reference is counted `unresolved` rather than read.
+ */
+function brokersFromChannelRefs(
+  channel: Record<string, unknown>,
+  root: Record<string, unknown>,
+): { brokers: Set<string>; explicit: boolean; unresolved: boolean } {
+  const out = new Set<string>();
+  const refs = own(channel, 'servers');
+  if (!Array.isArray(refs) || refs.length === 0) {
+    return { brokers: out, explicit: false, unresolved: false };
+  }
+  const servers = asRecord(own(root, 'servers'));
+  let unresolved = false;
+  for (const entry of refs) {
+    const ref = asString(own(entry, '$ref'));
+    const name = ref === undefined ? undefined : pointerName(ref, '#/servers/');
+    const target = name === undefined ? undefined : own(servers, name);
+    if (target === undefined) {
+      unresolved = true;
+      continue;
+    }
+    const resolved = serverBroker(target, root);
+    if (resolved.unresolved) {
+      unresolved = true;
+      continue;
+    }
+    if (resolved.broker !== undefined) out.add(resolved.broker);
+  }
+  return { brokers: out, explicit: true, unresolved };
+}
+
+/**
+ * Every broker the document's servers name, computed ONCE per document.
+ *
+ * A channel that names no `servers` is available on all of them — the
+ * specification's own default, not an inference. Computing it per operation was
+ * quadratic in `servers × operations`, which an in-cap document can drive to
+ * minutes.
+ */
+function brokersOfAllServers(root: Record<string, unknown>): {
+  brokers: Set<string>;
+  capped: boolean;
+  unresolved: boolean;
+} {
+  const out = new Set<string>();
+  const servers = asRecord(own(root, 'servers'));
+  if (servers === undefined) return { brokers: out, capped: false, unresolved: false };
+  let seen = 0;
+  for (const name in servers) {
+    if (!Object.prototype.hasOwnProperty.call(servers, name)) continue;
+    seen += 1;
+    if (seen > MAX_SERVERS_PER_DOCUMENT) {
+      // Inherited resolution refuses a capped map before asking it to agree
+      // with itself, so the brokers of the first thousand entries are unused.
+      return { brokers: out, capped: true, unresolved: false };
+    }
+  }
+  let unresolved = false;
+  for (const name in servers) {
+    if (!Object.prototype.hasOwnProperty.call(servers, name)) continue;
+    const resolved = serverBroker(own(servers, name), root);
+    if (resolved.unresolved) unresolved = true;
+    else if (resolved.broker !== undefined) out.add(resolved.broker);
+  }
+  return { brokers: out, capped: false, unresolved };
+}
+
+/**
+ * Root `asyncapi` version → readable, refused, or not a document at all.
+ *
+ * Compared on the MAJOR component only. A 3.1 document adds fields this module
+ * does not read and changes none it does; refusing it would lose real
+ * destinations over a minor-version digit.
+ */
+function classifyVersion(raw: Record<string, unknown>): 'read' | AsyncApiRefusal {
+  const declared = asString(own(raw, 'asyncapi'))?.trim();
+  if (declared === undefined || declared === '') return 'not-a-document';
+  const major = declared.split('.')[0];
+  if (major === '3') return 'read';
+  if (major === '2') return 'asyncapi-2-unsupported';
+  return 'unsupported-version';
+}
+
+export interface NormalizedDocument {
+  operations: AsyncApiOperation[];
+  refusals: Partial<Record<AsyncApiRefusal, number>>;
+  /** Operations EXAMINED, which is what the caps count. */
+  examined: number;
+  /** A bound stopped this document short. */
+  truncated: boolean;
+}
+
+/**
+ * Normalize one parsed document. Pure — no filesystem, so the whole refusal
+ * surface is testable from inline document literals.
+ *
+ * `budget` is the number of operations the RUN may still examine.
+ */
+export function normalizeAsyncApiDocument(
+  parsed: unknown,
+  documentPath: string,
+  budget: number = MAX_TOTAL_OPERATIONS,
+): NormalizedDocument {
+  const refusals: Partial<Record<AsyncApiRefusal, number>> = {};
+  const tally = makeTally(refusals);
+  const operations: AsyncApiOperation[] = [];
+  let examined = 0;
+  let truncated = false;
+
+  const raw = asRecord(parsed);
+  if (raw === undefined) {
+    tally.count('unparsable');
+    return { operations, refusals, examined, truncated };
+  }
+
+  const verdict = classifyVersion(raw);
+  if (verdict !== 'read') {
+    tally.count(verdict);
+    return { operations, refusals, examined, truncated };
+  }
+
+  const channels = asRecord(own(raw, 'channels'));
+  const operationsRaw = asRecord(own(raw, 'operations'));
+  if (operationsRaw === undefined) return { operations, refusals, examined, truncated };
+
+  const allServers = brokersOfAllServers(raw);
+  if (allServers.capped) {
+    tally.count('server-cap');
+    truncated = true;
+  }
+
+  for (const operationId of Object.keys(operationsRaw)) {
+    if (examined >= MAX_OPERATIONS_PER_DOCUMENT) {
+      tally.count('operation-cap');
+      truncated = true;
+      break;
+    }
+    if (examined >= budget) {
+      tally.count('total-operation-cap');
+      truncated = true;
+      break;
+    }
+    examined += 1;
+
+    const operation = asRecord(own(operationsRaw, operationId));
+    if (operation === undefined) {
+      tally.count('unparsable');
+      continue;
+    }
+
+    if (operationId.length > MAX_OPERATION_ID_LENGTH) {
+      tally.count('operation-id-too-long');
+      continue;
+    }
+
+    const action = asString(own(operation, 'action'))?.trim().toLowerCase();
+    if (action !== 'send' && action !== 'receive') {
+      tally.count('unrecognized-action');
+      continue;
+    }
+
+    const ref = asString(own(own(operation, 'channel'), '$ref'));
+    const channelName = ref === undefined ? undefined : pointerName(ref, '#/channels/');
+    if (channelName === undefined) {
+      tally.count('no-channel-reference');
+      continue;
+    }
+    const channel = asRecord(own(channels, channelName));
+    if (channel === undefined) {
+      tally.count('channel-not-found');
+      continue;
+    }
+    if (own(channel, '$ref') !== undefined && own(channel, 'address') === undefined) {
+      tally.count('unresolved-channel-reference');
+      continue;
+    }
+
+    // The `address` field, not the channel KEY. A generator is free to key a
+    // channel by anything unique; only `address` is defined as the thing the
+    // broker is addressed by, and keying a node on a document-local map key
+    // would join two services that merely organized their documents alike.
+    //
+    // NOT TRIMMED, deliberately. The source cascade keeps an address exactly as
+    // written — `" orders "` is its own node and does not join `"orders"` — on
+    // the grounds that a missing connection beats a false one. Two producers of
+    // one key must not hold opposite whitespace policies, and of the two
+    // available answers this is the one that errs away from joining.
+    const address = asString(own(channel, 'address'));
+    if (address === undefined || address.trim() === '') {
+      tally.count('no-address');
+      continue;
+    }
+    if (address.length > MAX_ADDRESS_LENGTH) {
+      tally.count('address-too-long');
+      continue;
+    }
+    // A non-empty `parameters` map is the specification's own statement that
+    // the address is a template. An EMPTY one states nothing — generators emit
+    // empty containers routinely — so it must not refuse a literal address; the
+    // `{` test below covers documents that template without declaring.
+    const parameters = asRecord(own(channel, 'parameters'));
+    if ((parameters !== undefined && Object.keys(parameters).length > 0) || address.includes('{')) {
+      tally.count('templated-address');
+      continue;
+    }
+
+    // BINDINGS FIRST. They are the operation's own statement about its broker;
+    // the servers are the channel's. Unioning every server before consulting
+    // the bindings made a document that declares both a REST server and a Kafka
+    // server lose every operation it states, filed under a reason that says the
+    // document contradicts itself — when the contradiction was manufactured
+    // here by asking a question the operation had already answered.
+    //
+    // The CHANNEL's bindings count as well. They are a statement about the same
+    // operation made one level up, and a conformant document may carry only
+    // those — `channels: { orders: { bindings: { kafka: {} } } }` with no
+    // operation binding and no usable server protocol was dropped as
+    // `protocol-unknown` while the document had said plainly which broker it
+    // meant. Where both levels speak and disagree, the document contradicts
+    // itself and neither answer may be used.
+    const fromBindings = brokersFromBindings(own(operation, 'bindings'));
+    for (const broker of brokersFromBindings(own(channel, 'bindings'))) {
+      fromBindings.add(broker);
+    }
+    if (fromBindings.size > 1) {
+      tally.count('protocol-disagreement');
+      continue;
+    }
+    const bindingBroker = [...fromBindings][0];
+
+    const explicitServers = brokersFromChannelRefs(channel, raw);
+    let broker: string | undefined;
+    if (bindingBroker !== undefined) {
+      // Cross-check only against servers the channel named itself, and only
+      // when they are unanimous. An inherited multi-protocol server set is not
+      // a claim about THIS operation.
+      const explicitBroker =
+        explicitServers.brokers.size === 1 ? [...explicitServers.brokers][0] : undefined;
+      if (explicitBroker !== undefined && explicitBroker !== bindingBroker) {
+        tally.count('protocol-disagreement');
+        continue;
+      }
+      broker = bindingBroker;
+    } else if (explicitServers.explicit) {
+      if (explicitServers.unresolved) {
+        tally.count('unresolved-server-reference');
+        continue;
+      }
+      if (explicitServers.brokers.size > 1) {
+        tally.count('protocol-disagreement');
+        continue;
+      }
+      broker = [...explicitServers.brokers][0];
+    } else {
+      // Order matters: an INCOMPLETE set must be refused before it is asked
+      // whether it agrees, because a subset agrees with itself for free.
+      if (allServers.capped) {
+        tally.count('capped-server-default');
+        continue;
+      }
+      if (allServers.unresolved) {
+        tally.count('unresolved-server-reference');
+        continue;
+      }
+      if (allServers.brokers.size > 1) {
+        tally.count('ambiguous-server-default');
+        continue;
+      }
+      broker = [...allServers.brokers][0];
+    }
+
+    if (broker === undefined) {
+      tally.count('protocol-unknown');
+      continue;
+    }
+    if (isNonDestinationBroker(broker)) {
+      tally.count('not-a-destination-protocol');
+      continue;
+    }
+
+    operations.push({ documentPath, operationId, action, address, broker });
+  }
+
+  return { operations, refusals, examined, truncated };
+}
+
+/**
+ * Cheap pre-parse gate: does this file even claim to be an AsyncAPI document?
+ *
+ * Scans the WHOLE text, which is already bounded by {@link MAX_DOCUMENT_BYTES}
+ * and already in memory. A fixed window is the wrong shape of bound here: it
+ * decides the answer by where the key happens to sit rather than by whether the
+ * key is there, so any window is a false negative waiting for a file with a
+ * longer preamble. Sixty-four kilobytes replaced four for exactly that reason
+ * and inherited exactly that defect — a licence header, a `$schema` block and a
+ * long `info.description` clear it easily. The gate exists to skip the YAML
+ * PARSE, which is the expensive half; a linear scan of the same bytes is not.
+ */
+function looksLikeDocument(text: string): boolean {
+  if (!text.includes('asyncapi')) return false;
+  return /(^|[\s{,"'])["']?asyncapi["']?\s*:/m.test(text);
+}
+
+interface WalkResult {
+  files: string[];
+  symlinksSkipped: number;
+  truncated: boolean;
+  unreadableDirectories: number;
+}
+
+async function collectCandidateFiles(root: string): Promise<WalkResult> {
+  const files: string[] = [];
+  let symlinksSkipped = 0;
+  let unreadableDirectories = 0;
+  let visited = 0;
+  let truncated = false;
+  // Distinct from `truncated`: a GLOBAL budget is exhausted and no further work
+  // is useful, whereas depth exhaustion in one branch says nothing about its
+  // siblings. Conflating them made a single over-deep subdirectory discard
+  // every remaining document in the walk, with the outcome decided by
+  // alphabetical ordering — a strictly worse failure than the one the
+  // truncation reporting was added to fix.
+  let exhausted = false;
+
+  const walk = async (dir: string, depth: number): Promise<void> => {
+    if (exhausted) return;
+    if (depth > MAX_DIRECTORY_DEPTH) {
+      truncated = true;
+      return;
+    }
+    let entries: import('node:fs').Dirent[];
+    try {
+      entries = await fs.readdir(dir, { withFileTypes: true });
+    } catch {
+      unreadableDirectories += 1;
+      truncated = true;
+      return;
+    }
+    // Sorted so the operation order a run produces is a function of the tree,
+    // not of the order the filesystem happened to hand entries back.
+    entries.sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0));
+    for (const entry of entries) {
+      if (exhausted) return;
+      visited += 1;
+      if (visited > MAX_WALK_ENTRIES || files.length >= MAX_DOCUMENTS) {
+        truncated = true;
+        exhausted = true;
+        return;
+      }
+      const full = path.join(dir, entry.name);
+      // `withFileTypes` reports a symlink as neither file nor directory, so
+      // links are skipped without ever being followed — a configured directory
+      // must not become a route out of itself. Counted rather than dropped in
+      // silence: a symlinked cache would otherwise look exactly like a wrong
+      // path.
+      if (entry.isSymbolicLink()) {
+        symlinksSkipped += 1;
+      } else if (entry.isDirectory()) {
+        await walk(full, depth + 1);
+      } else if (
+        entry.isFile() &&
+        DOCUMENT_EXTENSIONS.has(path.extname(entry.name).toLowerCase())
+      ) {
+        files.push(full);
+      }
+    }
+  };
+
+  await walk(root, 0);
+  return { files, symlinksSkipped, truncated, unreadableDirectories };
+}
+
+/**
+ * Read one candidate file under a hard byte ceiling.
+ *
+ * Modelled on `frameworks/spring/actuator-runtime.ts`'s `readPayloadFile`, and
+ * for the reason its comment gives: the size gate and the read share ONE handle
+ * so both observe the same inode. Checking `fs.stat(path)` and then re-resolving
+ * that path in `fs.readFile` lets whatever writes the directory swap the file
+ * between the two calls, which makes the cap advisory (CodeQL
+ * js/file-system-race). The out-of-band cache this option exists to read is
+ * written by other tooling by definition, so the race is the normal condition
+ * here rather than an exotic one.
+ *
+ * The read LOOPS, like its model. POSIX permits a short read on a regular file,
+ * and a single read was measured never short across seven hundred reads on
+ * APFS — but the deployments this option targets put the cache on NFS, SMB or a
+ * FUSE mount, and FUSE filesystems using `direct_io` do return short counts.
+ * The consequence of one short read is silent: a document truncated at a line
+ * boundary still parses, so operations vanish with `refusals: {}` and
+ * `truncated: false`, indistinguishable from a document that had fewer.
+ *
+ * The `isFile` test on the same handle is what the path-based version could not
+ * do at all. Without it the single-file configuration accepts anything `stat`
+ * follows: a character device reports size 0 and then streams until Node throws
+ * at two gigabytes, and a FIFO never returns at all.
+ *
+ * `O_NONBLOCK` is what makes that test reachable, and it was not obvious — it
+ * was found by writing the FIFO test and watching it TIME OUT rather than fail.
+ * Opening a FIFO for reading blocks in `open(2)` until some writer opens the
+ * other end, so a type check performed after the open never runs: the analyze
+ * hangs there, holding its repository lock, with no error to report. The flag
+ * makes the open return immediately for a FIFO and is a no-op for the regular
+ * files this actually wants (measured: identical byte count, digest and timing
+ * with and without it), which is why it costs nothing to keep.
+ */
+async function readBoundedFile(file: string): Promise<string | 'oversized' | 'unreadable'> {
+  let handle: import('node:fs/promises').FileHandle | undefined;
+  try {
+    // `O_NONBLOCK` is absent on some platforms; falling back to a plain
+    // read-only open there keeps behaviour identical for regular files.
+    const nonBlocking = (fsConstants.O_NONBLOCK ?? 0) | fsConstants.O_RDONLY;
+    handle = await fs.open(file, nonBlocking);
+    const stat = await handle.stat();
+    if (!stat.isFile()) return 'unreadable';
+    if (stat.size > MAX_DOCUMENT_BYTES) return 'oversized';
+    const buffer = Buffer.alloc(MAX_DOCUMENT_BYTES + 1);
+    let bytesRead = 0;
+    while (bytesRead < buffer.length) {
+      const chunk = await handle.read(buffer, bytesRead, buffer.length - bytesRead, bytesRead);
+      if (chunk.bytesRead === 0) break;
+      bytesRead += chunk.bytesRead;
+    }
+    if (bytesRead > MAX_DOCUMENT_BYTES) return 'oversized';
+    return buffer.subarray(0, bytesRead).toString('utf-8');
+  } catch {
+    return 'unreadable';
+  } finally {
+    await handle?.close().catch(() => {});
+  }
+}
+
+/**
+ * Read every AsyncAPI 3.x document under an explicitly configured path.
+ *
+ * `configuredPath` is resolved against the repository root, so an absolute path
+ * to a cache populated out of band and a repo-relative directory of committed
+ * documents are both natural — the same shape `springActuatorPath` offers for
+ * Actuator snapshots. The READ is wider than that neighbour's, and the
+ * difference is worth stating rather than glossed as "the same contract": the
+ * Actuator loader probes five fixed filenames in one directory, while this
+ * walks recursively under the caps above and opens every candidate it finds.
+ *
+ * There is deliberately NO glob-based auto-discovery. Scanning a repository for
+ * anything that parses as a document would make every existing index grow nodes
+ * on its next run with nobody having asked for it.
+ */
+export async function readAsyncApiDocuments(
+  repoPath: string,
+  configuredPath: string,
+): Promise<AsyncApiReadResult> {
+  const refusals: Partial<Record<AsyncApiRefusal, number>> = {};
+  const tally = makeTally(refusals);
+  const operations: AsyncApiOperation[] = [];
+  let documentsScanned = 0;
+  let documentsAccepted = 0;
+  let symlinksSkipped = 0;
+  let truncated = false;
+  let examinedTotal = 0;
+
+  const root = path.resolve(repoPath, configuredPath);
+  let files: string[];
+  try {
+    const stat = await fs.stat(root);
+    if (stat.isDirectory()) {
+      const walked = await collectCandidateFiles(root);
+      files = walked.files;
+      symlinksSkipped = walked.symlinksSkipped;
+      truncated = walked.truncated;
+      for (let i = 0; i < walked.unreadableDirectories; i += 1) tally.count('directory-unreadable');
+      if (truncated && walked.unreadableDirectories === 0) tally.count('walk-truncated');
+    } else {
+      files = [root];
+    }
+  } catch {
+    tally.count('unreadable');
+    return {
+      operations,
+      documentsScanned,
+      documentsAccepted,
+      symlinksSkipped,
+      truncated,
+      refusals,
+    };
+  }
+
+  for (const file of files) {
+    documentsScanned += 1;
+    const content = await readBoundedFile(file);
+    if (content === 'oversized' || content === 'unreadable') {
+      tally.count(content);
+      continue;
+    }
+
+    const text = content.startsWith(BOM) ? content.slice(BOM.length) : content;
+
+    // Sniff before parsing. A configured directory may hold hundreds of
+    // unrelated YAML files, and parsing each one to discover it is not a
+    // document is the difference between a bounded cost and a per-file one.
+    if (!looksLikeDocument(text)) {
+      tally.count('not-a-document');
+      continue;
+    }
+
+    let parsed: unknown;
+    try {
+      parsed = yaml.load(text, { schema: DOCUMENT_SCHEMA });
+    } catch {
+      tally.count('unparsable');
+      continue;
+    }
+
+    const remaining = MAX_TOTAL_OPERATIONS - examinedTotal;
+    if (remaining <= 0) {
+      tally.count('total-operation-cap');
+      truncated = true;
+      break;
+    }
+
+    const result = normalizeAsyncApiDocument(parsed, file, remaining);
+    examinedTotal += result.examined;
+    if (result.truncated) truncated = true;
+    for (const [reason, count] of Object.entries(result.refusals)) {
+      refusals[reason as AsyncApiRefusal] = (refusals[reason as AsyncApiRefusal] ?? 0) + count;
+    }
+    if (result.operations.length > 0) documentsAccepted += 1;
+    operations.push(...result.operations);
+  }
+
+  return { operations, documentsScanned, documentsAccepted, symlinksSkipped, truncated, refusals };
+}

gitnexus/src/core/ingestion/asyncapi/protocol.ts

@@ -0,0 +1,207 @@
+/**
+ * AsyncAPI protocol name → broker identity, for `destinationNodeKey`.
+ *
+ * Deliberately OUTSIDE `frameworks/spring/`, like `destination-key.ts` and for
+ * the same reason: an AsyncAPI document is not a Spring artifact, and the
+ * broker it names has to be mintable by anything that reads one.
+ *
+ * ── TWO READERS, TWO RULES, AND WHY THEY ARE NOT THE SAME RULE ────────────
+ *
+ * A document states its protocol in two places, and they have opposite
+ * defaults:
+ *
+ *   `servers[].protocol` is a FIELD DECLARED TO HOLD A PROTOCOL. Whatever it
+ *   contains is the document's claim about its broker, including a protocol
+ *   this codebase has never heard of. {@link brokerForProtocol} therefore
+ *   passes an unrecognized value through as its own literal — an `mqtt` or
+ *   `nats` channel mints `mqtt <address>` and joins any other site that says
+ *   the same thing, instead of being dropped for the sake of a closed union it
+ *   was never going to fit. Refusing it would lose a destination the document
+ *   states plainly, to protect against a collision that cannot happen: an
+ *   unmapped protocol keys on its own name, so it can only meet a site that
+ *   named the same protocol.
+ *
+ *   A `bindings` MAP KEY is not that. The map is keyed by protocol name BY
+ *   CONVENTION, and the specification puts other things in the same namespace:
+ *   `$ref` when the bindings are a Reference Object, and `x-` Specification
+ *   Extensions, which generators emit routinely. Here a non-protocol key is the
+ *   EXPECTED case, not the exotic one, so {@link brokerForBindingKey} answers
+ *   only for names it recognizes.
+ *
+ * That asymmetry was learned the expensive way. An earlier version applied one
+ * syntactic test to both and excluded only `$`-prefixed tokens; a document
+ * carrying `bindings: { x-scs-function: … }` then minted
+ * `Destination(broker='x-scs-function')`, so two unrelated services sharing a
+ * vendor annotation and an address landed on ONE node with a broker half that
+ * carried no broker information at all. Worse, `{ kafka: {}, x-internal: {} }`
+ * read as two brokers and refused a conformant document as self-contradictory
+ * — which also makes any writer of a document a one-line saboteur of its own
+ * cross-service links. A list that must be extended when AsyncAPI adds a
+ * binding is the smaller cost.
+ *
+ * ── WHY THE ALIASES EARN THEIR ROWS ───────────────────────────────────────
+ *
+ * `amqp` → `rabbit` is not an identity. AMQP is a wire protocol and RabbitMQ is
+ * one implementation of it; a Qpid or ActiveMQ broker speaking AMQP is filed
+ * under `rabbit` here and the label is wrong about the product. It is mapped
+ * anyway because the alternative is a guaranteed MISS: Spring's own capture
+ * calls `@RabbitListener` `rabbit`, so an `amqp` document describing the very
+ * same queue would sit on a second node and the two would never meet. A label
+ * that is wrong about the vendor but right about the protocol family joins the
+ * pair; an honest `amqp` label splits it every time.
+ *
+ * The transport-security variants are that argument with the vendor doubt
+ * removed. AsyncAPI's SERVER vocabulary distinguishes `kafka` from
+ * `kafka-secure`; its BINDINGS vocabulary does not. Without these rows a
+ * secured cluster's own document contradicts itself, and with bindings absent
+ * it is worse and quieter: `kafka-secure <address>` never meets the
+ * `kafka <address>` Spring capture mints, and nothing reports the miss. TLS is
+ * a property of the connection, not of the place messages go.
+ */
+
+/**
+ * A protocol name long enough to be a mistake.
+ *
+ * The broker is the THIRD string that reaches a graph identifier, alongside the
+ * address and the operation id, and it is the one that was left unbounded:
+ * `destinationNodeKey` is `` `${broker} ${address}` `` and `generateId` is
+ * `` `${label}:${name}` `` — concatenation both, no hashing. A one-megabyte
+ * protocol in a document that satisfies every other cap was measured producing
+ * a gigabyte of resident identifier strings, because the phase mints one id per
+ * node and one per edge. The longest name in AsyncAPI's vocabulary is
+ * `googlepubsub` at twelve characters, so this bound is generous by more than a
+ * factor of two and can only be reached on purpose.
+ */
+const MAX_PROTOCOL_LENGTH = 32;
+
+/**
+ * Spellings that differ between AsyncAPI's protocol vocabulary and the broker
+ * names this codebase already mints from source.
+ *
+ * Both AMQP versions collapse: `amqp1` is AMQP 1.0, a different wire format for
+ * the same family, and a service that documents one while its code speaks the
+ * other is describing one queue, not two. `mqtt5` collapses onto `mqtt` for the
+ * identical reason.
+ */
+const PROTOCOL_ALIASES: ReadonlyMap<string, string> = new Map([
+  ['amqp', 'rabbit'],
+  ['amqp1', 'rabbit'],
+  ['kafka-secure', 'kafka'],
+  ['secure-mqtt', 'mqtt'],
+  ['mqtts', 'mqtt'],
+  ['mqtt5', 'mqtt'],
+  ['wss', 'ws'],
+  ['stomps', 'stomp'],
+  ['https', 'http'],
+]);
+
+/**
+ * AsyncAPI's binding vocabulary — the names a `bindings` map key may take.
+ *
+ * Closed on purpose; see the header. Adding a protocol here is a deliberate
+ * act, which is the point: the cost of a missing row is one document's
+ * destinations, and the cost of an open door is a node keyed on a vendor
+ * annotation that two unrelated services happen to share.
+ */
+const BINDING_PROTOCOLS: ReadonlySet<string> = new Set([
+  'amqp',
+  'amqp1',
+  'anypointmq',
+  'googlepubsub',
+  'http',
+  'https',
+  'ibmmq',
+  'jms',
+  'kafka',
+  'kafka-secure',
+  'mercure',
+  'mqtt',
+  'mqtt5',
+  'mqtts',
+  'nats',
+  'pulsar',
+  'redis',
+  'secure-mqtt',
+  'sns',
+  'solace',
+  'sqs',
+  'stomp',
+  'stomps',
+  'ws',
+  'wss',
+]);
+
+/**
+ * Protocols whose destinations this module refuses to mint, because the address
+ * alone is not the thing that identifies them.
+ *
+ * For a broker, the topic or queue name IS the namespace: two services naming
+ * `orders.v1` on Kafka are talking about one place, and dropping which cluster
+ * they used is a bounded, stated trade. For HTTP and WebSocket the HOST is the
+ * namespace and the address is only a path, so keying on the path alone makes
+ * every service that exposes `/events` — or `/health`, or `/api/v1/orders` —
+ * one node. That is unbounded, and it is a false join rather than a lost one.
+ *
+ * These are not lost information: an HTTP endpoint is a `Route`, which the
+ * routes phase already models with the method in its key.
+ */
+const NON_DESTINATION_PROTOCOLS: ReadonlySet<string> = new Set(['http', 'ws']);
+
+/**
+ * Shape a protocol NAME must take, applied to both readers.
+ *
+ * The pass-through in {@link brokerForProtocol} is an argument about
+ * UNRECOGNIZED protocols — a name this codebase has not heard of is still the
+ * document's claim. It is not an argument about arbitrary text. A protocol name
+ * contains no whitespace, and one that did would collide in the node key, since
+ * `destinationNodeKey` joins broker and address with a space: `("kafka orders",
+ * "x")` and `("kafka", "orders x")` are then the same node.
+ *
+ * Learned twice. The check was added when that collision was first shown to be
+ * reachable, then dropped during a rewrite that moved the binding-key filtering
+ * into its own function — and the test written for the first lesson caught the
+ * second within the minute.
+ */
+function isProtocolToken(value: string): boolean {
+  return /^[a-z0-9][a-z0-9+._-]*$/.test(value);
+}
+
+function normalize(protocol: string | undefined): string | undefined {
+  if (protocol === undefined) return undefined;
+  const trimmed = protocol.trim().toLowerCase();
+  if (trimmed === '' || trimmed.length > MAX_PROTOCOL_LENGTH) return undefined;
+  if (!isProtocolToken(trimmed)) return undefined;
+  return trimmed;
+}
+
+/** True when a broker names a transport whose addresses must not be keyed. */
+export function isNonDestinationBroker(broker: string): boolean {
+  return NON_DESTINATION_PROTOCOLS.has(broker);
+}
+
+/**
+ * Normalize a `servers[].protocol` value to the broker half of a `Destination`
+ * key. Unrecognized protocols pass through; see the header.
+ *
+ * Returns `undefined` for a blank or implausibly long value — silence is not a
+ * claim, and a key built from an empty string would merge every silent
+ * document.
+ */
+export function brokerForProtocol(protocol: string | undefined): string | undefined {
+  const normalized = normalize(protocol);
+  if (normalized === undefined) return undefined;
+  return PROTOCOL_ALIASES.get(normalized) ?? normalized;
+}
+
+/**
+ * Normalize a `bindings` MAP KEY to a broker, answering only for names in
+ * AsyncAPI's binding vocabulary.
+ *
+ * `$ref` and `x-` extensions live in this namespace legitimately, so anything
+ * unrecognized is silence rather than a broker.
+ */
+export function brokerForBindingKey(key: string | undefined): string | undefined {
+  const normalized = normalize(key);
+  if (normalized === undefined || !BINDING_PROTOCOLS.has(normalized)) return undefined;
+  return PROTOCOL_ALIASES.get(normalized) ?? normalized;
+}

gitnexus/src/core/ingestion/frameworks/spring/destinations.ts

@@ -281,6 +281,31 @@ export interface SpringDestinationResolvers {
    * order is fixed now rather than renegotiated later. Nothing supplies it
    * today, so step 4 is a no-op and such destinations stay unresolved with the
    * reason the earlier step recorded.
+   *
+   * ── STILL UNSUPPLIED, AND NOW FOR A REASON RATHER THAN FOR WANT OF A READER ──
+   *
+   * `core/ingestion/asyncapi/document.ts` reads AsyncAPI 3.x documents, and
+   * `pipeline-phases/spring-destinations.ts` emits what they state as
+   * destinations of their own. It does NOT feed this hook, and the gap is a
+   * decision:
+   *
+   * A document names addresses; it does not name the method that uses one. To
+   * hand an address to THIS candidate, something has to choose which of the
+   * document's operations belongs to it. Partitioning by (broker, action) is
+   * the only division both sides agree on, and it is a weak one: a service with
+   * several listeners on one broker puts them all in one bucket. Any bucket
+   * holding more than one operation forces a heuristic, and a wrong heuristic
+   * puts a REAL address on a joining node under the wrong site — a false
+   * connection wearing the clothes of a resolved one, which is the exact
+   * outcome this module's keying rule exists to prevent. Only a bucket of size
+   * one is a fact rather than a guess.
+   *
+   * Two things would change that, and neither is a heuristic: a document whose
+   * operations carry the implementing symbol, or a configuration source that
+   * answers the `${key}` this candidate already recorded. The second is the
+   * stronger of the two — a key-to-value lookup is exact where a document match
+   * is a guess — and it wants its own resolver rather than this one, because
+   * what it needs is the placeholder key, not the candidate.
    */
   readonly specification?: (candidate: SpringDestinationCandidate) => string | null;
 }

gitnexus/src/core/ingestion/pipeline-phases/spring-destinations.ts

@@ -7,6 +7,15 @@
  * inbound and outbound facts are captured during parse and survive the parse
  * cache; until now nothing read them.
  *
+ * Since `asyncApiSpecPath`, the phase has a SECOND source that is not Spring
+ * and not source code at all: AsyncAPI documents read off disk
+ * (`ingestion/asyncapi/document.ts`). They mint the same `Destination` nodes on
+ * the same key, so both sources meet on one node. The reader is deliberately
+ * framework-neutral and lives outside `frameworks/spring/`; only the emit is
+ * hosted here, because this is where the node's keying rule is enforced and
+ * splitting that rule across two phases is how it drifts. The phase name is
+ * accurate about its origin rather than its current contents.
+ *
  * Shaped after `Route` + `HANDLES_ROUTE` in `routes.ts` — a framework overlay
  * node keyed by what it names, with the callable pointing at it, down to the
  * detail that the key pairs the address with the one dimension that can make
@@ -56,12 +65,16 @@
  * prevent.
  *
  * @deps    parse, scopeResolution, springConfig
- * @reads   Spring messaging capture facts, Method/Function nodes, Property nodes
- * @writes  Destination nodes; CONSUMES_FROM / PUBLISHES_TO / USES edges
+ * @reads   Spring messaging capture facts, Method/Function nodes, Property nodes,
+ *          AsyncAPI documents under `options.asyncApiSpecPath` (filesystem)
+ * @writes  Destination nodes; synthetic File nodes for out-of-tree documents;
+ *          CONSUMES_FROM / PUBLISHES_TO / USES edges
  */
 
+import path from 'node:path';
 import type { GraphNode, Range } from 'gitnexus-shared';
 import { generateId } from '../../../lib/utils.js';
+import { readAsyncApiDocuments } from '../asyncapi/document.js';
 import { logger } from '../../logger.js';
 import type { KnowledgeGraph } from '../../graph/types.js';
 import { SPRING_CONFIG_DESCRIPTION } from '../frameworks/spring/config-bindings.js';
@@ -101,6 +114,37 @@ export interface SpringDestinationsOutput {
   readonly refusalsByReason: Readonly<Record<string, number>>;
   /** Destination -> Property provenance edges for `${key}` placeholders. */
   readonly configKeyLinks: number;
+  /**
+   * What reading AsyncAPI documents contributed, present only when
+   * `asyncApiSpecPath` was configured. Absent means "not asked for", which is
+   * deliberately distinguishable from a configured path that yielded nothing —
+   * a mistyped directory and a repository with no documents are different
+   * problems with different fixes, and one zero cannot say which happened.
+   */
+  readonly specDocuments?: SpecDocumentStats;
+}
+
+export interface SpecDocumentStats {
+  /** Entries skipped because they were symbolic links, and whether a bound
+   *  stopped the walk. Both make every other number here a FLOOR, and a floor
+   *  reported as a total is the failure this whole block exists to prevent. */
+  readonly symlinksSkipped: number;
+  readonly truncated: boolean;
+  /** Files considered under the configured path. */
+  readonly scanned: number;
+  /** Files that parsed as an AsyncAPI 3.x document and yielded an operation. */
+  readonly accepted: number;
+  /** Operations normalized to a (broker, address, action) triple. */
+  readonly operations: number;
+  /** Destination nodes this reading minted that no source site had already. */
+  readonly destinations: number;
+  /** CONSUMES_FROM + PUBLISHES_TO edges from documents. */
+  readonly edges: number;
+  /** Document- and operation-level refusals, by reason. Kept apart from
+   *  `refusalsByReason` above: that number is the denominator of the SOURCE
+   *  unresolved fraction, and a mistyped specification directory must not be
+   *  able to make the source look worse than it is. */
+  readonly refusalsByReason: Readonly<Record<string, number>>;
 }
 
 /**
@@ -274,6 +318,192 @@ function edgeReason(candidate: SpringDestinationCandidate): string {
   return `spring-${candidate.source}:${element}${exchange}`;
 }
 
+/**
+ * Pseudo-path prefix for a document that is not a file of this repository.
+ *
+ * An edge needs a source node, and the emit below refuses to attach one to a
+ * `File` that does not exist — so a document supplied from outside the working
+ * tree needs an identity minted for it. The same answer
+ * `frameworks/spring/actuator-runtime.ts` gives for Actuator snapshots
+ * (`spring-actuator:<endpoint>`): a prefixed pseudo-path that a real
+ * repo-relative path is not expected to take.
+ *
+ * That is a CONVENTION, not a guarantee, and the difference is worth stating
+ * because the neighbouring prefix states it too strongly. A colon is a legal
+ * POSIX filename character, so a committed file literally named
+ * `asyncapi:orders.yaml` would share this identity — costing one merged node
+ * and a misattributed edge, never a wrong address. Windows cannot express the
+ * collision at all. It is accepted on the same terms the Actuator prefix
+ * already is rather than escaped, because an escape would have to be applied to
+ * both prefixes at once to be worth anything.
+ */
+const DOCUMENT_FILE_PREFIX = 'asyncapi:';
+
+/**
+ * The `File` node an AsyncAPI operation's edge hangs off.
+ *
+ * A document COMMITTED to the repository already has a real `File` node, and
+ * using it is strictly better: the edge lands on something the reader can open,
+ * and the ordinary per-file writeback keeps it honest. Only a document from
+ * outside the tree gets a synthetic node.
+ */
+function documentFileNodeId(
+  ctx: PipelineContext,
+  configuredRoot: string,
+  documentPath: string,
+): string {
+  const repoRelative = path.relative(ctx.repoPath, documentPath);
+  if (repoRelative !== '' && !repoRelative.startsWith('..') && !path.isAbsolute(repoRelative)) {
+    const realId = generateId('File', repoRelative.split(path.sep).join('/'));
+    if (ctx.graph.getNode(realId) !== undefined) return realId;
+  }
+  // Relative to the CONFIGURED root, not to the filesystem root: an absolute
+  // path would put a machine's directory layout into the graph, and two
+  // machines indexing the same documents would then disagree about their ids.
+  const relative = path.relative(configuredRoot, documentPath);
+  const label =
+    relative === '' || relative.startsWith('..')
+      ? path.basename(documentPath)
+      : relative.split(path.sep).join('/');
+  const filePath = `${DOCUMENT_FILE_PREFIX}${label}`;
+  const id = generateId('File', filePath);
+  if (ctx.graph.getNode(id) === undefined) {
+    ctx.graph.addNode({
+      id,
+      label: 'File',
+      properties: { name: path.basename(documentPath), filePath },
+    });
+  }
+  return id;
+}
+
+/**
+ * Mint destinations stated by AsyncAPI documents, with no claim about code.
+ *
+ * ── WHY THIS DOES NOT TRY TO FIND THE HANDLER ─────────────────────────────
+ *
+ * A document says an address is sent to or received from; it does not say by
+ * which method. Guessing that mapping is a real temptation and a bad trade: the
+ * addresses in one document partition by (broker, action) into buckets that
+ * usually hold more than one operation, so any assignment beyond a bucket of
+ * size one is a heuristic — and a wrong one silently attaches a real address to
+ * the wrong handler, which is a false connection dressed as a resolved one.
+ *
+ * So this claims only what the document actually states: that THIS SERVICE
+ * talks to that address on that broker in that direction. The edge therefore
+ * starts at the document, not at a callable. That is a weaker statement than a
+ * source-derived edge and it is worth having anyway, because it is available in
+ * cases where the source cannot supply one at all — a listener registered
+ * programmatically, a broker this codebase has no patterns for, or a language
+ * whose messaging idiom nobody has taught it yet.
+ *
+ * The node itself is the ordinary resolved `Destination`: same key, same
+ * `address` property, so a document and a source site that name one address on
+ * one broker land on ONE node and the two halves of a conversation meet. That
+ * is the whole point, and it is why this mints nothing of its own invention.
+ */
+async function emitSpecDestinations(
+  ctx: PipelineContext,
+  specPath: string,
+): Promise<SpecDocumentStats> {
+  const read = await readAsyncApiDocuments(ctx.repoPath, specPath);
+  const configuredRoot = path.resolve(ctx.repoPath, specPath);
+  let destinations = 0;
+  let edges = 0;
+
+  for (const operation of read.operations) {
+    const nodeId = generateId(
+      'Destination',
+      destinationNodeKey(operation.broker, operation.address),
+    );
+    // Runs AFTER the source pass, so a site that resolved this address already
+    // owns the node and keeps its own `resolution` provenance. First writer
+    // wins and the order is fixed, so the property is deterministic rather than
+    // a race — and `literal` is the more informative of the two answers anyway.
+    if (ctx.graph.getNode(nodeId) === undefined) {
+      ctx.graph.addNode({
+        id: nodeId,
+        label: 'Destination',
+        properties: {
+          name: operation.address,
+          // Empty for the same reason every connecting destination carries it
+          // empty: the node is shared, and stamping it with the document's path
+          // would make it collateral damage of that path's next writeback.
+          filePath: '',
+          address: operation.address,
+          // NOT `'specification'`, though the address did come from one. That
+          // value belongs to `SpringDestinationVia` and means "a CODE
+          // CANDIDATE was resolved through the step-4 resolver hook" — a
+          // different fact with a code site behind it. Reusing it would make a
+          // query that groups destinations by provenance unable to separate an
+          // address a document merely states from one a document was used to
+          // resolve, and the second of those is a claim about source that this
+          // node is not making.
+          resolution: 'asyncapi-document',
+          broker: operation.broker,
+        },
+      });
+      destinations += 1;
+    }
+
+    const sourceId = documentFileNodeId(ctx, configuredRoot, operation.documentPath);
+    const type = operation.action === 'receive' ? 'CONSUMES_FROM' : 'PUBLISHES_TO';
+    const reason = `asyncapi:${operation.operationId}`;
+    ctx.graph.addRelationship({
+      id: generateId(type, `${sourceId}->${nodeId}:${reason}`),
+      sourceId,
+      targetId: nodeId,
+      type,
+      confidence: 1.0,
+      reason,
+    });
+    edges += 1;
+  }
+
+  const stats: SpecDocumentStats = {
+    symlinksSkipped: read.symlinksSkipped,
+    truncated: read.truncated,
+    scanned: read.documentsScanned,
+    accepted: read.documentsAccepted,
+    operations: read.operations.length,
+    destinations,
+    edges,
+    refusalsByReason: read.refusals as Readonly<Record<string, number>>,
+  };
+
+  // Unconditional, and not `isDev`-gated like the summary below it. The tally
+  // above is justified on the grounds that an operator must be able to tell a
+  // mistyped directory from a repository with no documents — and that
+  // justification is only true if the operator can SEE it. A configured path
+  // that produced nothing is the one outcome where silence and success look
+  // identical from outside, which is why `spring-auto-configuration.ts` warns
+  // unconditionally for the same class of input.
+  if (stats.accepted === 0 || stats.truncated) {
+    // Repo-relative when the path is inside the repository, bare name when it
+    // is not. The same change refuses to persist this path to index metadata on
+    // the grounds that it would record an operator's directory layout; applying
+    // that reasoning to metadata and not to logs would be holding one rule in
+    // two places.
+    const resolved = path.resolve(ctx.repoPath, specPath);
+    const relative = path.relative(ctx.repoPath, resolved);
+    const reportedPath =
+      relative !== '' && !relative.startsWith('..') && !path.isAbsolute(relative)
+        ? relative.split(path.sep).join('/')
+        : path.basename(resolved);
+    logger.warn(
+      {
+        asyncApiSpecPath: reportedPath,
+        ...stats,
+      },
+      stats.accepted === 0
+        ? '⚠️ No AsyncAPI document under the configured path yielded a destination.'
+        : '⚠️ AsyncAPI document reading hit a bound; the destinations below are a floor, not a total.',
+    );
+  }
+
+  return stats;
+}
+
 export const springDestinationsPhase: PipelinePhase<SpringDestinationsOutput> = {
   name: 'springDestinations',
   // `parse` supplies the file list and the harvested constants; `scopeResolution`
@@ -362,13 +592,28 @@ export const springDestinationsPhase: PipelinePhase<SpringDestinationsOutput> =
         );
       }
     }
+    // Documents are read whether or not the source pass found anything, and
+    // that is the point of the closure rather than a straight call here: a
+    // repository whose messaging is invisible to the source patterns — a broker
+    // with no rules, a listener registered programmatically — is exactly the
+    // case a published document exists to cover, and an early return keyed on
+    // source sites would skip the documents precisely there.
+    //
+    // Always invoked AFTER the source emit, so a site that resolved an address
+    // owns its node first and keeps its own provenance.
+    const specPath = ctx.options?.asyncApiSpecPath;
+    const readSpecifications = async (): Promise<SpecDocumentStats | undefined> =>
+      specPath === undefined ? undefined : emitSpecDestinations(ctx, specPath);
+
     if (sites.length === 0) {
+      const specDocuments = await readSpecifications();
       return {
         resolvedDestinations: 0,
         unresolvedDestinations: 0,
         edges: 0,
         refusalsByReason,
         configKeyLinks: 0,
+        ...(specDocuments === undefined ? {} : { specDocuments }),
       };
     }
 
@@ -548,9 +793,20 @@ export const springDestinationsPhase: PipelinePhase<SpringDestinationsOutput> =
       edges += 1;
     }
 
+    const specDocuments = await readSpecifications();
+
     if (isDev) {
+      const fromSpec =
+        specDocuments === undefined
+          ? ''
+          : `, +${specDocuments.destinations} from ${specDocuments.accepted} document(s)`;
+      // The breakdown, not just the totals. The unresolved FRACTION is the
+      // number this feature is judged on, and a bare count of unresolved
+      // destinations says how big the gap is without saying what would close
+      // it — which is the only question an operator can act on.
       logger.info(
-        `📮 Spring destinations: ${resolvedDestinations} resolved, ${unresolvedDestinations} unresolved, ${edges} edges`,
+        { refusalsByReason, ...(specDocuments === undefined ? {} : { specDocuments }) },
+        `📮 Spring destinations: ${resolvedDestinations} resolved, ${unresolvedDestinations} unresolved, ${edges} edges${fromSpec}`,
       );
     }
 
@@ -560,6 +816,7 @@ export const springDestinationsPhase: PipelinePhase<SpringDestinationsOutput> =
       edges,
       refusalsByReason,
       configKeyLinks,
+      ...(specDocuments === undefined ? {} : { specDocuments }),
     };
   },
 };

gitnexus/src/core/ingestion/pipeline.ts

@@ -74,6 +74,20 @@ export interface PipelineOptions {
   springActuatorPath?: string;
   /** Repo-relative Actuator inputs retained only for a cleanup scan. */
   springActuatorScanExclusions?: readonly string[];
+  /**
+   * Explicit local AsyncAPI 3.x document input, read by the `springDestinations`
+   * phase. Accepts a directory of documents or a single document; the path is
+   * resolved against the repository root, so a committed `docs/asyncapi` and an
+   * absolute cache populated out of band are equally natural. Undefined keeps
+   * specification reading completely disabled.
+   *
+   * There is deliberately no glob-based auto-discovery to go with it. Scanning
+   * a repository for anything that parses as a document would make every
+   * existing index grow destination nodes on its next run without an operator
+   * having decided anything — the same reason new contract extractors ship
+   * opt-in rather than on.
+   */
+  asyncApiSpecPath?: string;
   /** Per-advice Spring AOP candidate inspection cap. `0` disables this cap. */
   springAopMaxCandidateInspectionsPerAdvice?: number;
   /** Aggregate Spring AOP candidate inspection cap for one analysis. `0` disables this cap. */

gitnexus/src/core/run-analyze.ts

@@ -469,6 +469,11 @@ export interface AnalyzeOptions {
    * the Spring enrichment phase. Undefined keeps static-only analysis.
    */
   springActuatorPath?: string;
+  /**
+   * Explicit local AsyncAPI 3.x document input, forwarded to the destination
+   * phase. Undefined keeps source-only address resolution.
+   */
+  asyncApiSpecPath?: string;
   /**
    * The caller will `process.exit()` immediately after this analyze returns (the
    * CLI `analyze` command). When set, the finalize/error close CHECKPOINTs for
@@ -1708,6 +1713,35 @@ async function runFullAnalysisInner(
   const springActuatorScanExclusions =
     retainedActuatorInputs.length === 0 ? undefined : retainedActuatorInputs;
 
+  // AsyncAPI documents are the same class of input as Actuator snapshots and
+  // need the same treatment, for a reason git cannot see: the documents live
+  // outside the tree as often as in it, and NOTHING about replacing one moves
+  // the commit or dirties the working tree. Without this, the second run of an
+  // out-of-band cache — the workflow the option exists for — takes the
+  // already-up-to-date fast path below, never opens a document, and serves the
+  // previous run's addresses while reporting success. Measured, not reasoned:
+  // editing a document and re-running printed "Already up to date" and left the
+  // old address in the graph.
+  //
+  // Forcing the rebuild also settles a second defect for free. A synthetic
+  // `File` node for an out-of-tree document (`asyncapi:<label>`) carries a path
+  // that is in no write set and is not covered by `isGraphWideNode`, so on an
+  // incremental writeback the node is dropped while its edges — anchored to a
+  // graph-wide `Destination` — are kept, and the edges then COPY against a row
+  // that was never written. A full rebuild has no incremental subgraph to get
+  // that wrong, so the pair cannot come apart.
+  const asyncApiSpecRequested = options.asyncApiSpecPath !== undefined;
+  const asyncApiSpecPreviouslyEnabled = existingMeta?.asyncApiSpec?.enabled === true;
+  if (asyncApiSpecRequested) {
+    if (!options.force) {
+      log('AsyncAPI document reading requested; forcing a full rebuild.');
+    }
+    options = { ...options, force: true };
+  } else if (asyncApiSpecPreviouslyEnabled) {
+    log('AsyncAPI document reading disabled; rebuilding to remove document-derived evidence.');
+    options = { ...options, force: true };
+  }
+
   // ── Early-return: already up to date ──────────────────────────────
   if (
     existingMeta &&
@@ -2015,6 +2049,7 @@ async function runFullAnalysisInner(
       fetchWrappers: options.fetchWrappers,
       skipDerivedGraphPhases,
       springActuatorPath: options.springActuatorPath,
+      asyncApiSpecPath: options.asyncApiSpecPath,
       springActuatorScanExclusions,
     },
   );
@@ -3797,6 +3832,11 @@ async function runFullAnalysisInner(
             },
           }
         : {}),
+      // Written only while enabled. Once the option is dropped, the disable
+      // transition above has already forced the cleanup rebuild, so carrying a
+      // `{ enabled: false }` stamp forward would only make every subsequent run
+      // re-decide a question that is already settled.
+      ...(asyncApiSpecRequested ? { asyncApiSpec: { enabled: true } } : {}),
       // Branch identity this index represents (#2106). Recorded for the flat
       // slot too (so resolveBranchPlacement knows which branch owns it). When
       // the label is null (detached HEAD / non-git re-analyze) we PRESERVE an

gitnexus/src/mcp/resources.ts

@@ -368,6 +368,13 @@ async function getContextResource(backend: LocalBackend, repoName?: string): Pro
   lines.push(`  runner_identity: ${JSON.stringify(freshMeta?.runnerIdentity ?? null)}`);
   lines.push(`  incomplete_reasons: ${JSON.stringify(incompleteReasons)}`);
   lines.push(`  spring_actuator: ${JSON.stringify(freshMeta?.springActuator ?? null)}`);
+  // Surfaced beside the Actuator flag, and it matters more than that one does:
+  // Actuator only annotates nodes the source pass already found, whereas
+  // document reading MINTS destinations and edges that have no code site at
+  // all. Without this line an agent reading the context sees `Destination`
+  // nodes rooted at `asyncapi:`-prefixed pseudo-files with nothing to say where
+  // they came from.
+  lines.push(`  asyncapi_spec: ${JSON.stringify(freshMeta?.asyncApiSpec ?? null)}`);
   const indexedRunnerSchema = (freshMeta?.runnerIdentity as { schemaVersion?: unknown } | undefined)
     ?.schemaVersion;
   lines.push(

gitnexus/src/server/analyze-launch.ts

@@ -47,6 +47,7 @@ export interface LaunchOptions {
   embeddings?: boolean;
   dropEmbeddings?: boolean;
   springActuatorPath?: string;
+  asyncApiSpecPath?: string;
   registryName?: string;
 }
 
@@ -336,6 +337,7 @@ export function createLaunchAnalysisWorker(deps: LaunchDeps) {
           embeddings: !!opts.embeddings,
           dropEmbeddings: !!opts.dropEmbeddings,
           ...(opts.springActuatorPath ? { springActuatorPath: opts.springActuatorPath } : {}),
+          ...(opts.asyncApiSpecPath ? { asyncApiSpecPath: opts.asyncApiSpecPath } : {}),
           ...(opts.registryName ? { registryName: opts.registryName } : {}),
         },
       });

gitnexus/src/server/api.ts

@@ -1521,6 +1521,7 @@ export const createServer = async (port: number, host: string = '127.0.0.1') =>
           embeddings,
           dropEmbeddings,
           springActuatorPath,
+          asyncApiSpecPath,
           token: repoToken,
         } = req.body;
 
@@ -1540,6 +1541,13 @@ export const createServer = async (port: number, host: string = '127.0.0.1') =>
           res.status(400).json({ error: '"springActuatorPath" must be a non-empty string' });
           return;
         }
+        if (
+          asyncApiSpecPath !== undefined &&
+          (typeof asyncApiSpecPath !== 'string' || asyncApiSpecPath.trim().length === 0)
+        ) {
+          res.status(400).json({ error: '"asyncApiSpecPath" must be a non-empty string' });
+          return;
+        }
 
         if (!repoUrl && !repoLocalPath) {
           res.status(400).json({ error: 'Provide "url" (git URL) or "path" (local path)' });
@@ -1628,6 +1636,7 @@ export const createServer = async (port: number, host: string = '127.0.0.1') =>
               embeddings,
               dropEmbeddings,
               springActuatorPath,
+              asyncApiSpecPath,
             });
           } catch (err: any) {
             if (targetPath) releaseRepoLock(getStoragePath(targetPath));

gitnexus/src/storage/repo-meta.ts

@@ -98,6 +98,23 @@ export interface RepoMeta {
      */
     repoRelativeInputs: string[];
   };
+  /**
+   * Whether the index was built with AsyncAPI document reading enabled.
+   *
+   * Only the FLAG is recorded, unlike `springActuator` above, because the two
+   * options need different things from this field. Actuator inputs are retained
+   * so future scans keep excluding them; documents are deliberately NOT
+   * excluded from scanning — a committed one wants its real `File` node — so
+   * there is nothing to retain, and recording the configured path would put an
+   * operator's absolute directory layout into index metadata for no consumer.
+   *
+   * The flag alone is what the disable transition needs: without it, dropping
+   * the option leaves document-derived destinations in an index with nothing
+   * able to notice they should go.
+   */
+  asyncApiSpec?: {
+    enabled: boolean;
+  };
   /**
    * Analyzer/runtime receipt for the successful run represented by this
    * metadata. Optional so indexes written by older GitNexus releases remain

gitnexus/test/unit/analyze-worker-pool-size.test.ts

@@ -107,6 +107,27 @@ describe('analyzeCommand --workers validation', () => {
     );
   });
 
+  it('threads --asyncapi-spec through to the core pipeline', async () => {
+    // Without this the whole flag is inert-but-green: the option parses, the
+    // phase reads a property nobody sets, and every test still passes because
+    // the phase tests hand-build their own context.
+    const { analyzeCommand } = await import('../../src/cli/analyze.js');
+    runFullAnalysisMock.mockResolvedValue({
+      repoName: 'repo',
+      repoPath: '/repo',
+      stats: {},
+      alreadyUpToDate: true,
+    });
+
+    await analyzeCommand(undefined, { asyncapiSpec: 'docs/asyncapi' });
+
+    expect(runFullAnalysisMock).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.objectContaining({ asyncApiSpecPath: 'docs/asyncapi' }),
+      expect.any(Object),
+    );
+  });
+
   it('rejects --workers 0 with a CLI error (sequential parsing was removed)', async () => {
     const { analyzeCommand } = await import('../../src/cli/analyze.js');
     runFullAnalysisMock.mockResolvedValue({

gitnexus/test/unit/incremental-orchestration.test.ts

@@ -1817,3 +1817,128 @@ describe('runFullAnalysis — escalated wipe recreates the vector index (#2409,
     }
   }, 600_000);
 });
+
+/** Document-derived destinations currently in the graph, address + provenance. */
+async function readDocumentDestinations(
+  repoPath: string,
+): Promise<Array<{ address: string; broker: string; resolution: string }>> {
+  const adapter = await import('../../src/core/lbug/lbug-adapter.js');
+  const { lbugPath } = getStoragePaths(repoPath);
+  await adapter.initLbug(lbugPath);
+  try {
+    const rows = (await adapter.executeQuery(
+      `MATCH (d:Destination) WHERE d.resolution = 'asyncapi-document' ` +
+        `RETURN d.address AS address, d.broker AS broker, d.resolution AS resolution ` +
+        `ORDER BY address`,
+    )) as Array<{ address?: unknown; broker?: unknown; resolution?: unknown }>;
+    return rows.map((row) => ({
+      address: String(row.address ?? ''),
+      broker: String(row.broker ?? ''),
+      resolution: String(row.resolution ?? ''),
+    }));
+  } finally {
+    await adapter.closeLbug();
+  }
+}
+
+describe('runFullAnalysis — AsyncAPI document reading', () => {
+  /**
+   * Drives the REAL `runFullAnalysis` with `asyncApiSpecPath` set.
+   *
+   * Three separate things were individually deletable with the whole suite
+   * green before this existed: the forward from `run-analyze` into
+   * `PipelineOptions`, the forced rebuild while the option is enabled, and the
+   * cleanup rebuild when it is dropped. Each of them makes the feature
+   * partially or wholly inert, and none of them is visible one layer up, where
+   * the CLI test asserts on a mock's arguments.
+   *
+   * The document lives OUTSIDE the repository on purpose. That is the workflow
+   * the option is documented for — a cache written by other tooling — and it is
+   * the only one where the hazard is real: editing a tracked file dirties the
+   * tree and would force a rebuild anyway, so an in-repo fixture would pass
+   * even with the freshness fix reverted.
+   */
+  it('forces a rebuild while enabled, re-reads a changed document, and cleans up once on disable', async () => {
+    const repo = await setupMiniRepo();
+    // A SIBLING of the repository, not a child: the document must be outside
+    // the working tree, or editing it would dirty the tree and force a rebuild
+    // on its own, and the test would pass with the freshness fix reverted.
+    const specDir = path.join(repo.dbPath, '..', 'gnx-asyncapi-spec');
+    await mkdir(specDir, { recursive: true });
+    const specFile = path.join(specDir, 'orders.yaml');
+    const documentFor = (address: string): string =>
+      [
+        'asyncapi: 3.0.0',
+        'info: { title: Order Service, version: 1.0.0 }',
+        'servers: { broker: { host: "example:9092", protocol: kafka } }',
+        `channels: { c: { address: ${address}, servers: [{ $ref: "#/servers/broker" }] } }`,
+        'operations: { publishOrder: { action: send, channel: { $ref: "#/channels/c" } } }',
+        '',
+      ].join('\n');
+
+    try {
+      await writeFile(specFile, documentFor('orders.v1'), 'utf-8');
+      const { runFullAnalysis } = await import('../../src/core/run-analyze.js');
+      const { storagePath } = getStoragePaths(repo.dbPath);
+
+      const enabledLogs: string[] = [];
+      const enabled = await runFullAnalysis(
+        repo.dbPath,
+        { skipAgentsMd: true, asyncApiSpecPath: specDir },
+        { onProgress: () => {}, onLog: (message) => enabledLogs.push(message) },
+      );
+      expect(enabled.alreadyUpToDate).toBeUndefined();
+      expect(enabledLogs.join('\n')).toContain(
+        'AsyncAPI document reading requested; forcing a full rebuild.',
+      );
+      // The forward into PipelineOptions is what puts this node in the graph;
+      // without it the flag parses and nothing else happens.
+      expect(await readDocumentDestinations(repo.dbPath)).toEqual([
+        { address: 'orders.v1', broker: 'kafka', resolution: 'asyncapi-document' },
+      ]);
+      expect((await loadMeta(storagePath))?.asyncApiSpec).toEqual({ enabled: true });
+
+      // Same commit, clean tree, only the out-of-tree document changed. Git
+      // freshness cannot see this, so without the forced rebuild the run is a
+      // no-op that reports success and serves the previous address.
+      await writeFile(specFile, documentFor('orders.v2'), 'utf-8');
+      const rereadRun = await runFullAnalysis(
+        repo.dbPath,
+        { skipAgentsMd: true, asyncApiSpecPath: specDir },
+        { onProgress: () => {} },
+      );
+      expect(rereadRun.alreadyUpToDate).toBeUndefined();
+      expect(await readDocumentDestinations(repo.dbPath)).toEqual([
+        { address: 'orders.v2', broker: 'kafka', resolution: 'asyncapi-document' },
+      ]);
+
+      const disableLogs: string[] = [];
+      const disabled = await runFullAnalysis(
+        repo.dbPath,
+        { skipAgentsMd: true },
+        { onProgress: () => {}, onLog: (message) => disableLogs.push(message) },
+      );
+      expect(disabled.alreadyUpToDate).toBeUndefined();
+      expect(disableLogs.join('\n')).toContain(
+        'AsyncAPI document reading disabled; rebuilding to remove document-derived evidence.',
+      );
+      expect(await readDocumentDestinations(repo.dbPath)).toEqual([]);
+      expect((await loadMeta(storagePath))?.asyncApiSpec).toBeUndefined();
+
+      // Exactly ONE cleanup rebuild: the metadata write drops the flag, so the
+      // next run has nothing to react to. A merge-over-previous write here
+      // would rebuild forever with nothing failing.
+      const steady = await runFullAnalysis(
+        repo.dbPath,
+        { skipAgentsMd: true },
+        { onProgress: () => {} },
+      );
+      expect(steady.alreadyUpToDate).toBe(true);
+    } finally {
+      // The document directory is a sibling of the repository, so the repo's
+      // own cleanup does not reach it — both owners have to be called here.
+      await rm(specDir, { recursive: true, force: true });
+      await repo.cleanup();
+    }
+  }, 180_000);
+});

gitnexus/test/unit/shipped-skills-sync.test.ts

@@ -120,6 +120,12 @@ describe('intended standard-skill improvements stay in every applicable copy', (
     }
   });
 
+  it('documents the AsyncAPI analyze flag in every CLI copy', () => {
+    for (const file of standardSkillCopies('gitnexus-cli')) {
+      expect(fs.readFileSync(file, 'utf-8')).toContain('`--asyncapi-spec <path>`');
+    }
+  });
+
   // These copies are NOT byte-compared (only the engineering FAMILY above is),
   // so a runner added to resolve-analyze-cmd.cjs can silently miss them. The
   // audience that most needs bunx documented — a bun-only machine with no npm,

gitnexus/test/unit/spring-destinations-phase.test.ts

@@ -1,5 +1,9 @@
-import { beforeEach, describe, expect, it } from 'vitest';
+import { afterAll, beforeEach, describe, expect, it } from 'vitest';
+import { mkdtemp, writeFile, symlink, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
 import { createKnowledgeGraph } from '../../src/core/graph/graph.js';
+import { _captureLogger } from '../../src/core/logger.js';
 import {
   setJavaSpringMessageProducerFacts,
   setJavaSpringNonHttpHandlerFacts,
@@ -629,3 +633,298 @@ describe('destinationNodeKey', () => {
     expect(destinationNodeKey('', 'orders')).toBe('orders');
   });
 });
+
+/**
+ * AsyncAPI documents as a second source of destinations.
+ *
+ * DIRECTION IS THE ASSERTION THAT MATTERS. A swapped send/receive mapping
+ * emits both edge types, both nodes, and a fully connected graph — only with
+ * every arrow reversed. Any test that asserts a node or an edge EXISTS passes
+ * identically under the broken mapping, so each test below names the edge TYPE
+ * it expects for a given action.
+ */
+describe('springDestinations phase — AsyncAPI documents', () => {
+  let graph: KnowledgeGraph;
+
+  beforeEach(() => {
+    graph = createKnowledgeGraph();
+  });
+
+  const KAFKA_DOCUMENT = `
+asyncapi: 3.0.0
+info: { title: Order Service, version: 1.0.0 }
+servers:
+  broker: { host: "example:9092", protocol: kafka }
+channels:
+  outbound:
+    address: orders
+    servers: [{ $ref: "#/servers/broker" }]
+  inbound:
+    address: shipments
+    servers: [{ $ref: "#/servers/broker" }]
+operations:
+  publishOrder:
+    action: send
+    channel: { $ref: "#/channels/outbound" }
+  onShipment:
+    action: receive
+    channel: { $ref: "#/channels/inbound" }
+`;
+
+  // Tracked and removed; an earlier version of this suite left hundreds of
+  // temporary directories behind on developer machines.
+  const createdDirs: string[] = [];
+  afterAll(async () => {
+    for (const dir of createdDirs) await rm(dir, { recursive: true, force: true });
+  });
+
+  async function specDir(body: string = KAFKA_DOCUMENT): Promise<string> {
+    const dir = await mkdtemp(path.join(tmpdir(), 'gnx-phase-spec-'));
+    createdDirs.push(dir);
+    await writeFile(path.join(dir, 'asyncapi.yaml'), body, 'utf-8');
+    return dir;
+  }
+
+  async function runWithSpec(
+    files: string[],
+    asyncApiSpecPath: string | undefined,
+  ): Promise<SpringDestinationsOutput> {
+    const deps = new Map<string, PhaseResult<unknown>>([
+      [
+        'parse',
+        {
+          phaseName: 'parse',
+          durationMs: 0,
+          output: { allPaths: files, moduleConstants: new Map() },
+        },
+      ],
+      ['scopeResolution', { phaseName: 'scopeResolution', durationMs: 0, output: {} }],
+      ['springConfig', { phaseName: 'springConfig', durationMs: 0, output: {} }],
+    ]);
+    const ctx = {
+      repoPath: '/repo',
+      graph,
+      onProgress: () => {},
+      pipelineStart: 0,
+      ...(asyncApiSpecPath === undefined ? {} : { options: { asyncApiSpecPath } }),
+    } as unknown as PipelineContext;
+    return springDestinationsPhase.execute(ctx, deps) as Promise<SpringDestinationsOutput>;
+  }
+
+  function edgesFrom(address: string): { type: string; sourceId: string }[] {
+    const target = generateId('Destination', destinationNodeKey('kafka', address));
+    return [...graph.iterRelationships()]
+      .filter((r) => r.targetId === target)
+      .map((r) => ({ type: r.type, sourceId: r.sourceId }));
+  }
+
+  it('maps `send` to PUBLISHES_TO and `receive` to CONSUMES_FROM', async () => {
+    const output = await runWithSpec([], await specDir());
+
+    expect(output.specDocuments?.operations).toBe(2);
+    expect(output.specDocuments?.destinations).toBe(2);
+    expect(output.specDocuments?.edges).toBe(2);
+
+    // Swapping the mapping would leave both of these arrays non-empty and both
+    // nodes present; only the TYPE distinguishes the two readings.
+    expect(edgesFrom('orders').map((e) => e.type)).toEqual(['PUBLISHES_TO']);
+    expect(edgesFrom('shipments').map((e) => e.type)).toEqual(['CONSUMES_FROM']);
+  });
+
+  it('gives a spec-minted destination NO file path, and its own provenance', async () => {
+    await runWithSpec([], await specDir());
+    const node = [...graph.iterNodes()].find(
+      (n) => n.id === generateId('Destination', destinationNodeKey('kafka', 'orders')),
+    );
+    // `filePath: ''` is load-bearing, not cosmetic: a connecting destination is
+    // shared by every site that names it, and the incremental writeback deletes
+    // by file. Stamping it with the document's path would make a shared node
+    // collateral damage of that document's next change, taking every OTHER
+    // referrer's edge with it via DETACH DELETE.
+    expect(node?.properties.filePath).toBe('');
+    // Distinct from `'specification'`, which belongs to the address cascade and
+    // means a CODE candidate was resolved through the step-4 hook — a claim
+    // about source that this node is not making.
+    expect(node?.properties.resolution).toBe('asyncapi-document');
+  });
+
+  it('reads documents even when the source pass found no messaging at all', async () => {
+    // The early return used to be keyed on source sites alone. A repository
+    // whose broker this codebase has no patterns for is exactly the case a
+    // published document covers, so skipping documents there would drop the
+    // feature where it is most needed.
+    const output = await runWithSpec([], await specDir());
+    expect(output.resolvedDestinations).toBe(0);
+    expect(output.specDocuments?.destinations).toBe(2);
+  });
+
+  it('lands a document and a source site on ONE node for one (broker, address)', async () => {
+    const filePath = 'src/Orders.java';
+    setJavaSpringNonHttpHandlerFacts(filePath, []);
+    setJavaSpringMessageProducerFacts(filePath, [
+      {
+        ownerScopeId: `${filePath}#publish` as never,
+        ownerRange: OWNER_RANGE,
+        template: 'kafka',
+        receiverName: 'kafkaTemplate',
+        methodName: 'send',
+        args: [{ text: '"orders"' }, { text: 'payload' }],
+      },
+    ]);
+    callableNode(graph, filePath, 'publish');
+
+    const output = await runWithSpec([filePath], await specDir());
+
+    const nodeId = generateId('Destination', destinationNodeKey('kafka', 'orders'));
+    const nodes = [...graph.iterNodes()].filter((n) => n.id === nodeId);
+    expect(nodes).toHaveLength(1);
+    expect(nodes[0].properties.address).toBe('orders');
+    // The source pass ran first and owns the provenance; the document joined
+    // its node rather than minting a second one.
+    expect(nodes[0].properties.resolution).toBe('literal');
+    expect(output.resolvedDestinations).toBe(1);
+    expect(output.specDocuments?.destinations).toBe(1);
+    // Two edges into one node — the publisher's and the document's — which is
+    // the whole point: both halves of a conversation meet on one node. Asserted
+    // by TYPE rather than by count: a count of two survives a reversal in
+    // either of the two paths that produced them.
+    expect(edgesFrom('orders').map((e) => e.type)).toEqual(['PUBLISHES_TO', 'PUBLISHES_TO']);
+  });
+
+  it('hangs the edge off the document’s REAL File node when it is in the repo', async () => {
+    // The in-repo branch is the one the code calls "strictly better", and it is
+    // the branch that decides whether the graph grows a permanent pseudo-File
+    // node per document. Without a test, deleting it is invisible.
+    const dir = await specDir();
+    // `repoPath` is the fixture directory, so the document resolves inside it.
+    const deps = new Map<string, PhaseResult<unknown>>([
+      [
+        'parse',
+        { phaseName: 'parse', durationMs: 0, output: { allPaths: [], moduleConstants: new Map() } },
+      ],
+      ['scopeResolution', { phaseName: 'scopeResolution', durationMs: 0, output: {} }],
+      ['springConfig', { phaseName: 'springConfig', durationMs: 0, output: {} }],
+    ]);
+    const ctx = {
+      repoPath: dir,
+      graph,
+      onProgress: () => {},
+      pipelineStart: 0,
+      options: { asyncApiSpecPath: 'asyncapi.yaml' },
+    } as unknown as PipelineContext;
+    // The document sits at <dir>/asyncapi.yaml, so its repo-relative path is
+    // `asyncapi.yaml`; register that File node and expect the edge to use it.
+    const inRepoId = generateId('File', 'asyncapi.yaml');
+    graph.addNode({
+      id: inRepoId,
+      label: 'File',
+      properties: { name: 'asyncapi.yaml', filePath: 'asyncapi.yaml' },
+    });
+    await springDestinationsPhase.execute(ctx, deps);
+    const [edge] = edgesFrom('orders');
+    expect(edge.sourceId).toBe(inRepoId);
+    expect(edge.sourceId).not.toBe(generateId('File', 'asyncapi:asyncapi.yaml'));
+  });
+
+  it('never lets a document give an unresolved destination an address', async () => {
+    const filePath = 'src/Placeholder.java';
+    setJavaSpringNonHttpHandlerFacts(filePath, [
+      {
+        ownerScopeId: `${filePath}#consume` as never,
+        ownerFilePath: filePath,
+        ownerRange: OWNER_RANGE,
+        annotations: [
+          { name: 'KafkaListener', args: [{ name: 'topics', text: '"${app.topic}"' }] },
+        ],
+      },
+    ]);
+    setJavaSpringMessageProducerFacts(filePath, []);
+    callableNode(graph, filePath, 'consume');
+
+    const output = await runWithSpec([filePath], await specDir());
+
+    expect(output.unresolvedDestinations).toBe(1);
+    const unresolved = [...graph.iterNodes()].filter(
+      (n) => n.label === 'Destination' && n.properties.resolution === 'unresolved-config-key',
+    );
+    expect(unresolved).toHaveLength(1);
+    // The document names real addresses, and one of them may even be the value
+    // behind this placeholder — but nothing here knows that, so the node keeps
+    // its location key and stays unjoinable.
+    expect(unresolved[0].properties.address).toBeUndefined();
+  });
+
+  it('hangs the edge off a synthetic File when the document is outside the repo', async () => {
+    await runWithSpec([], await specDir());
+    const [edge] = edgesFrom('orders');
+    const source = [...graph.iterNodes()].find((n) => n.id === edge.sourceId);
+    expect(source?.label).toBe('File');
+    expect(String(source?.properties.filePath)).toBe('asyncapi:asyncapi.yaml');
+  });
+
+  it('forwards a non-zero symlink count out of the reader', async () => {
+    // The phase's stats block is the operator's only view of what was read.
+    // Hard-coding these to zero passed every test, because the only assertion
+    // on them was an all-zeros comparison on an empty directory.
+    const dir = await specDir();
+    await symlink(path.join(dir, 'asyncapi.yaml'), path.join(dir, 'linked.yaml'));
+    const output = await runWithSpec([], dir);
+    expect(output.specDocuments?.symlinksSkipped).toBe(1);
+  });
+
+  it('warns, out loud, when a configured path yielded nothing', async () => {
+    // The stats block is justified on the grounds that an operator must be able
+    // to tell a mistyped directory from a repository with no documents. That is
+    // only true if the warn actually fires, and nothing asserted that it did —
+    // the block could be deleted with the suite green.
+    const empty = await mkdtemp(path.join(tmpdir(), 'gnx-phase-empty-'));
+    createdDirs.push(empty);
+    const populated = await specDir();
+
+    const capture = _captureLogger();
+    try {
+      await runWithSpec([], empty);
+      const warnings = capture
+        .records()
+        .filter((record) => (record as { level?: number }).level === 40);
+      expect(warnings).toHaveLength(1);
+      expect((warnings[0] as { accepted?: number }).accepted).toBe(0);
+    } finally {
+      capture.restore();
+    }
+
+    const quiet = _captureLogger();
+    try {
+      await runWithSpec([], populated);
+      expect(
+        quiet.records().filter((record) => (record as { level?: number }).level === 40),
+      ).toHaveLength(0);
+    } finally {
+      quiet.restore();
+    }
+  });
+
+  it('omits the stats block entirely when no path was configured', async () => {
+    // Absent must stay distinguishable from "configured and found nothing":
+    // a mistyped directory and a repository with no documents need different
+    // answers from an operator, and one zero cannot say which happened.
+    const output = await runWithSpec([], undefined);
+    expect(output.specDocuments).toBeUndefined();
+    expect('specDocuments' in output).toBe(false);
+  });
+
+  it('reports a configured path that yielded nothing, rather than staying silent', async () => {
+    const empty = await mkdtemp(path.join(tmpdir(), 'gnx-phase-spec-empty-'));
+    const output = await runWithSpec([], empty);
+    expect(output.specDocuments).toEqual({
+      symlinksSkipped: 0,
+      truncated: false,
+      scanned: 0,
+      accepted: 0,
+      operations: 0,
+      destinations: 0,
+      edges: 0,
+      refusalsByReason: {},
+    });
+  });
+});

gitnexus/test/unit/watch-paths.test.ts

@@ -139,6 +139,11 @@ describe('watch path selection', () => {
       [{ skipSkills: true }, '--skip-skills'],
       [{ stats: false }, '--no-stats'],
       [{ springActuator: './actuator' }, '--spring-actuator'],
+      // Rejected for the same reason as the Actuator path: the watcher reacts
+      // to source changes and nothing watches a document directory, so
+      // accepting the flag would read the documents once and then serve a
+      // stale answer for the rest of the session.
+      [{ asyncapiSpec: './docs/asyncapi' }, '--asyncapi-spec'],
     ];
     for (const [options, flag] of unsupportedCliOptions) {
       await expect(