From 0a1828726b5debd57b7583fe0132897d3ffd4a81 Mon Sep 17 00:00:00 2001 From: Michael Egner Date: Wed, 23 Sep 2026 20:06:32 +0200 Subject: [PATCH 1/2] v0.5.0 I4 slice 1: decision and canonical foundation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Record the I4 Pub/Sub gate outcome as GO (spec §4, §15 Slice 1): - docs/specifications/0.5.0/i4-decision-evidence.md: independent evidence record (ASB, GCP Pub/Sub, OTel, Kafka; retrieved 2026-09-23), §4 distinction mapping, §4.2 GO disposition, §19 checklist dispositions. - ADR 0017 (Proposed): supersedes ADR 0013 only for the Topic/Subscription prohibition; both guards retained. ADR 0013 status and index updated. Add the in-memory canonical foundation (no adapter behavior, no persisted Pub/Sub node or relation, spec §11): - canonical Topic/Subscription models (not yet on ArchitectureModel); - topic_owned_id / subscription_owned_id identity helpers (§7.1/§7.2); - the three §8.4 DiagnosticCode members (vocabulary only); - Topic/Subscription uniqueness constraints; - public contract skeleton: EntityType TOPIC/SUBSCRIPTION, EvidenceRelationType PUBLISHES_TO/SUBSCRIPTION_OF, DeliveryRelationType PUBLISHES_TO, DeliveryRef.subscription with §12.2 invariants mirrored in the regenerated v0.5 schemas. A fail-closed test pins that importer/registry/canonicalization-v2 still exclude Pub/Sub until slice 2 opens that path atomically. Co-Authored-By: Claude Opus 5.5 --- app/architecture_intelligence/contracts.py | 94 +++++++++-- app/canonical/model.py | 23 +++ app/graph/schema.py | 5 + app/sources/model.py | 9 ++ app/sources/owner_ids.py | 49 ++++++ .../0013-no-topic-family-without-guards.md | 2 + ...017-source-independent-pubsub-semantics.md | 78 ++++++++++ docs/adr/README.md | 3 +- .../0.5.0/i4-decision-evidence.md | 147 ++++++++++++++++++ .../v0.5/architecture-answer.schema.json | 126 ++++++++++++++- .../v0.5/drift-answer.schema.json | 126 ++++++++++++++- .../v0.5/evidence-answer.schema.json | 132 +++++++++++++++- tests/integration/test_importer.py | 2 + ...est_architecture_intelligence_contracts.py | 142 +++++++++++++++++ .../unit/test_canonical_pubsub_foundation.py | 51 ++++++ tests/unit/test_sources_model.py | 18 +++ tests/unit/test_sources_owner_ids.py | 106 +++++++++++++ 17 files changed, 1083 insertions(+), 30 deletions(-) create mode 100644 docs/adr/0017-source-independent-pubsub-semantics.md create mode 100644 docs/specifications/0.5.0/i4-decision-evidence.md create mode 100644 tests/unit/test_canonical_pubsub_foundation.py diff --git a/app/architecture_intelligence/contracts.py b/app/architecture_intelligence/contracts.py index b654af89..718b7b29 100644 --- a/app/architecture_intelligence/contracts.py +++ b/app/architecture_intelligence/contracts.py @@ -57,6 +57,9 @@ class EntityType(StrEnum): OPERATION = "OPERATION" QUEUE = "QUEUE" WORKLOAD = "WORKLOAD" + # v0.5.0 I4 spec §12.1. + TOPIC = "TOPIC" + SUBSCRIPTION = "SUBSCRIPTION" class WorkloadKind(StrEnum): @@ -76,6 +79,7 @@ class DeliveryKind(StrEnum): class DeliveryRelationType(StrEnum): CALLS = "CALLS" SENDS = "SENDS" + PUBLISHES_TO = "PUBLISHES_TO" # v0.5.0 I4 spec §12.2 class DestinationResolution(StrEnum): @@ -118,9 +122,10 @@ class DependencyPredicate(StrEnum): class EvidenceRelationType(StrEnum): - """v0.4.0 I2.1 - the 7 canonical graph relation kinds (spec §11.2's `supports`). Deliberately its - own closed enum rather than reusing `DeliveryRelationType` (only CALLS/SENDS) or a graph-layer - string - `get_evidence` describes existing facts, never a new architecture claim.""" + """v0.4.0 I2.1 - the closed set of canonical graph relation kinds (spec §11.2's `supports`), + widened by I3 (DEPLOYED_AS) and v0.5.0 I4 (PUBLISHES_TO/SUBSCRIPTION_OF). Deliberately its own + closed enum rather than reusing `DeliveryRelationType` (only the delivery relations) or a + graph-layer string - `get_evidence` describes existing facts, never a new architecture claim.""" PROVIDES = "PROVIDES" CALLS = "CALLS" @@ -130,14 +135,22 @@ class EvidenceRelationType(StrEnum): CONFORMS_TO = "CONFORMS_TO" DEAD_LETTERS_TO = "DEAD_LETTERS_TO" DEPLOYED_AS = "DEPLOYED_AS" # I3 spec §16.1 + # v0.5.0 I4 spec §12.1: exactly these two are added; RECEIVES_FROM and CARRIES are reused. + PUBLISHES_TO = "PUBLISHES_TO" + SUBSCRIPTION_OF = "SUBSCRIPTION_OF" # Fixed (kind, relation_type, via.type) pairs - spec §11.2/§13. No other combination is valid. _ALLOWED_DELIVERY_PAIRS = { (DeliveryKind.SYNC_HTTP, DeliveryRelationType.CALLS, EntityType.OPERATION), (DeliveryKind.ASYNC_MESSAGE, DeliveryRelationType.SENDS, EntityType.QUEUE), + # v0.5.0 I4 spec §12.2. + (DeliveryKind.ASYNC_MESSAGE, DeliveryRelationType.PUBLISHES_TO, EntityType.TOPIC), } +# v0.5.0 I4 spec §12.1: `EntityRef` may carry bounded protocol/namespace metadata for these only. +_DESTINATION_ENTITY_TYPES = frozenset({EntityType.QUEUE, EntityType.TOPIC, EntityType.SUBSCRIPTION}) + ProducerName = Literal["architecture-intelligence-platform"] PRODUCER_NAME: ProducerName = get_args(ProducerName)[0] @@ -207,7 +220,10 @@ def _entity_ref_schema_extra(schema: dict, _model: type[BaseModel]) -> None: "else": {"properties": {"method": {"type": "null"}, "path": {"type": "null"}}}, }, { - "if": {"properties": {"type": {"const": "QUEUE"}}, "required": ["type"]}, + "if": { + "properties": {"type": {"enum": sorted(_DESTINATION_ENTITY_TYPES)}}, + "required": ["type"], + }, "else": {"properties": {"protocol": {"type": "null"}, "namespace": {"type": "null"}}}, }, ] @@ -230,10 +246,12 @@ class EntityRef(BaseModel): def _check_type_specific_fields(self) -> EntityRef: if self.type != EntityType.OPERATION and (self.method is not None or self.path is not None): raise ValueError("method/path are only allowed when type == OPERATION") - if self.type != EntityType.QUEUE and ( + if self.type not in _DESTINATION_ENTITY_TYPES and ( self.protocol is not None or self.namespace is not None ): - raise ValueError("protocol/namespace are only allowed when type == QUEUE") + raise ValueError( + "protocol/namespace are only allowed when type is QUEUE/TOPIC/SUBSCRIPTION" + ) return self @@ -264,10 +282,10 @@ class WorkloadRef(BaseModel): def _delivery_ref_schema_extra(schema: dict, _model: type[BaseModel]) -> None: - """Encode the fixed (kind, relation_type, via.type) pairs table (spec §11.2/§13) as JSON - Schema if/then so external (non-Pydantic) validators reject the same invalid combinations. - The Python model_validator below is still authoritative at runtime - this only mirrors it - for the committed schema.""" + """Encode the fixed (kind, relation_type, via.type) pairs table (spec §11.2/§13, widened by + v0.5.0 I4 spec §12.2) and the I4 `subscription` invariants as JSON Schema if/then so external + (non-Pydantic) validators reject the same invalid combinations. The Python model_validator + below is still authoritative at runtime - this only mirrors it for the committed schema.""" schema["allOf"] = [ *schema.get("allOf", []), { @@ -282,13 +300,55 @@ def _delivery_ref_schema_extra(schema: dict, _model: type[BaseModel]) -> None: }, { "if": {"properties": {"kind": {"const": "ASYNC_MESSAGE"}}, "required": ["kind"]}, + "then": { + "properties": {"relation_type": {"enum": ["PUBLISHES_TO", "SENDS"]}}, + "required": ["relation_type", "via"], + }, + }, + { + "if": { + "properties": {"relation_type": {"const": "SENDS"}}, + "required": ["relation_type"], + }, "then": { "properties": { - "relation_type": {"const": "SENDS"}, "via": {"properties": {"type": {"const": "QUEUE"}}, "required": ["type"]}, }, - "required": ["relation_type", "via"], + "required": ["via"], + }, + }, + { + "if": { + "properties": {"relation_type": {"const": "PUBLISHES_TO"}}, + "required": ["relation_type"], + }, + "then": { + "properties": { + "via": {"properties": {"type": {"const": "TOPIC"}}, "required": ["type"]}, + }, + "required": ["via"], + }, + }, + # I4 spec §12.2: `subscription` is null unless `via` is a Topic, and a non-null + # `subscription` is a SUBSCRIPTION-typed ref. + { + "if": { + "properties": { + "via": {"properties": {"type": {"const": "TOPIC"}}, "required": ["type"]}, + }, + "required": ["via"], }, + "else": {"properties": {"subscription": {"type": "null"}}}, + }, + { + "properties": { + "subscription": { + "anyOf": [ + {"type": "null"}, + {"properties": {"type": {"const": "SUBSCRIPTION"}}, "required": ["type"]}, + ] + } + } }, ] @@ -301,6 +361,11 @@ class DeliveryRef(BaseModel): kind: DeliveryKind relation_type: DeliveryRelationType via: EntityRef + # v0.5.0 I4 spec §12.2: the optional Pub/Sub Subscription route. Always emitted (as null when + # absent), like `EntityRef`'s own optional fields. That the Subscription is `SUBSCRIPTION_OF` + # the `via` Topic in the same snapshot is a projection invariant (graph-dependent), not a + # model rule. + subscription: EntityRef | None = None @model_validator(mode="after") def _check_allowed_pair(self) -> DeliveryRef: @@ -309,6 +374,11 @@ def _check_allowed_pair(self) -> DeliveryRef: raise ValueError( f"unsupported delivery (kind, relation_type, via.type) combination: {pair}" ) + if self.subscription is not None: + if self.via.type != EntityType.TOPIC: + raise ValueError("subscription is only allowed when via.type == TOPIC") + if self.subscription.type != EntityType.SUBSCRIPTION: + raise ValueError("subscription.type must be SUBSCRIPTION") return self diff --git a/app/canonical/model.py b/app/canonical/model.py index 16292013..5fad6861 100644 --- a/app/canonical/model.py +++ b/app/canonical/model.py @@ -39,6 +39,29 @@ class Queue(BaseModel): queue_type: str = "STANDARD" +class Topic(BaseModel): + """v0.5.0 I4 spec §6.2: a publish destination whose downstream fan-out is expressed only by + distinct Subscriptions. Deliberately separate from `Queue` (no `queue_type`) and from any generic + destination supertype (ADR 0017). Not yet carried by `ArchitectureModel` - slice 2 adds it + together with the importer path and the canonicalization-v3 bump (spec §11).""" + + id: str + name: str + protocol: str | None = None + namespace: str | None = None + + +class Subscription(BaseModel): + """v0.5.0 I4 spec §6.2: a stable named logical delivery entity associated with exactly one Topic. + The Topic association is the `SUBSCRIPTION_OF` relation, not a field here. Carries no consumer + instances, consumer groups, partitions, offsets, lag, filters, or delivery guarantees.""" + + id: str + name: str + protocol: str | None = None + namespace: str | None = None + + class Message(BaseModel): id: str name: str diff --git a/app/graph/schema.py b/app/graph/schema.py index f42a3200..c6aab84f 100644 --- a/app/graph/schema.py +++ b/app/graph/schema.py @@ -6,6 +6,11 @@ "CREATE CONSTRAINT service_id IF NOT EXISTS FOR (s:Service) REQUIRE s.id IS UNIQUE", "CREATE CONSTRAINT operation_id IF NOT EXISTS FOR (o:Operation) REQUIRE o.id IS UNIQUE", "CREATE CONSTRAINT queue_id IF NOT EXISTS FOR (q:Queue) REQUIRE q.id IS UNIQUE", + # v0.5.0 I4 slice 1 (spec §11/§15): uniqueness for the Topic/Subscription identities defined in + # §7. Schema only - no Topic/Subscription node is persisted until slice 2's canonicalization-v3 + # bump opens the importer path. + "CREATE CONSTRAINT topic_id IF NOT EXISTS FOR (t:Topic) REQUIRE t.id IS UNIQUE", + ("CREATE CONSTRAINT subscription_id IF NOT EXISTS FOR (s:Subscription) REQUIRE s.id IS UNIQUE"), "CREATE CONSTRAINT message_id IF NOT EXISTS FOR (m:Message) REQUIRE m.id IS UNIQUE", "CREATE CONSTRAINT schema_id IF NOT EXISTS FOR (s:Schema) REQUIRE s.id IS UNIQUE", "CREATE CONSTRAINT evidence_id IF NOT EXISTS FOR (e:Evidence) REQUIRE e.id IS UNIQUE", diff --git a/app/sources/model.py b/app/sources/model.py index d84b78fa..f89935c0 100644 --- a/app/sources/model.py +++ b/app/sources/model.py @@ -264,6 +264,15 @@ class DiagnosticCode(StrEnum): # error) - since `mappingId` is part of both §13.1's group-key formula and §8.3's evidence # identity, a silent collision here would be a latent identity defect, not a benign duplicate. SERVICE_WORKLOAD_MAPPING_DUPLICATE_ID = "SERVICE_WORKLOAD_MAPPING_DUPLICATE_ID" + # v0.5.0 I4 spec §8.4's own named codes - exactly these three, and no DESTINATION_KIND_UNSUPPORTED + # member. Vocabulary only as of I4 slice 1; the AsyncAPI Topic/Subscription mapping that emits + # them lands in slice 2. TOPIC_IDENTITY_CONFLICT: configured vs derived Topic id disagree. + # SUBSCRIPTION_IDENTITY_CONFLICT: configured Subscription id, Topic binding, or Subscription name + # disagrees with its declared/derived counterpart. SUBSCRIPTION_IDENTITY_MISSING: a Topic + # subscribe operation with no explicit or configured Subscription identity. + TOPIC_IDENTITY_CONFLICT = "TOPIC_IDENTITY_CONFLICT" + SUBSCRIPTION_IDENTITY_CONFLICT = "SUBSCRIPTION_IDENTITY_CONFLICT" + SUBSCRIPTION_IDENTITY_MISSING = "SUBSCRIPTION_IDENTITY_MISSING" class IngestionDiagnostic(BaseModel): diff --git a/app/sources/owner_ids.py b/app/sources/owner_ids.py index 0fb1b5fb..36e20564 100644 --- a/app/sources/owner_ids.py +++ b/app/sources/owner_ids.py @@ -142,3 +142,52 @@ def queue_owned_id( _utf8(exact_channel_address), ) return f"queue:owned:{sha256_hex(key)}" + + +def topic_owned_id( + *, + stable_broker_id: str, + normalized_namespace_or_empty: str, + exact_topic_address: str, +) -> str: + """v0.5.0 I4 spec §7.1: + + topic_owner_key = length-delimited( + stable broker id, normalized namespace-or-empty, exact normalized topic/channel address) + topic_id = topic:owned: + + Same inputs as `queue_owned_id`, but a distinct prefix - Queue and Topic ids never alias merely + because their inputs match. As with `queue_owned_id`, the caller applies Unicode NFC first. + """ + key = length_delimited( + _utf8(stable_broker_id), + _utf8(normalized_namespace_or_empty), + _utf8(exact_topic_address), + ) + return f"topic:owned:{sha256_hex(key)}" + + +def subscription_owned_id( + *, + stable_broker_id: str, + normalized_namespace_or_empty: str, + topic_id: str, + exact_subscription_name: str, +) -> str: + """v0.5.0 I4 spec §7.2: + + subscription_owner_key = length-delimited( + stable broker id, normalized namespace-or-empty, canonical Topic id, + exact normalized subscription name) + subscription_id = subscription:owned: + + Binding the canonical Topic id keeps identical Subscription names on different Topics distinct. + There is deliberately no consumer-group input (spec §7.2). The caller applies Unicode NFC first. + """ + key = length_delimited( + _utf8(stable_broker_id), + _utf8(normalized_namespace_or_empty), + _utf8(topic_id), + _utf8(exact_subscription_name), + ) + return f"subscription:owned:{sha256_hex(key)}" diff --git a/docs/adr/0013-no-topic-family-without-guards.md b/docs/adr/0013-no-topic-family-without-guards.md index 48817878..efb51331 100644 --- a/docs/adr/0013-no-topic-family-without-guards.md +++ b/docs/adr/0013-no-topic-family-without-guards.md @@ -2,6 +2,8 @@ Status: Accepted — promotes an existing `v0.3` cross-system decision into the ADR index; it does not re-decide it. +Superseded in part by [0017](0017-source-independent-pubsub-semantics.md): only the +Topic/Subscription prohibition is superseded, and both guards are retained. ## Context diff --git a/docs/adr/0017-source-independent-pubsub-semantics.md b/docs/adr/0017-source-independent-pubsub-semantics.md new file mode 100644 index 00000000..24a26957 --- /dev/null +++ b/docs/adr/0017-source-independent-pubsub-semantics.md @@ -0,0 +1,78 @@ +# 17. Source-independent Topic/Subscription semantics, with both ADR 0013 guards retained + +Status: Proposed. The `GO` decision was recorded in `v0.5.0` I4 Slice 1. This ADR moves to +Accepted when the I4 work it describes lands (I4 Slice 6). It supersedes +[ADR 0013](0013-no-topic-family-without-guards.md) **only** with respect to ADR 0013's +Topic/Subscription prohibition. + +## Context + +[ADR 0013](0013-no-topic-family-without-guards.md) prohibited a topic/pub-sub canonical family +until two guards existed. The first was a topic-vs-queue destination guard, and the second a +service-identity guard. It named the condition for lifting that prohibition: a new ADR citing the +guards' regression tests. + +`v0.4.1` I2 implemented both guards in the production runtime messaging path: +`decide_destination_semantics` and `decide_service_identity` in +`app/telemetry/messaging_guards.py`, evaluated by +`app/telemetry/adapter.py::correlate_queue_observations`. Their regression evidence is +`tests/unit/test_messaging_guards.py` (the D1–D17 and S1–S16 matrices), +`tests/unit/test_adapter.py` (the composed C1–C17 matrix and the Quarkus/Airflow reachability +proofs) and `tests/integration/test_adapter.py` (real-Neo4j zero-artifact refusal proofs). ADR +0013's own implementation record lists them. + +The `v0.5.0` parent specification made Pub/Sub conditional on a `GO`/`DEFER` gate. +[`i4-source-independent-pubsub-semantics.md`](../specifications/0.5.0/i4-source-independent-pubsub-semantics.md) +(Draft 0.3, merged at `f98e48b`) is the governing specification. +[`i4-decision-evidence.md`](../specifications/0.5.0/i4-decision-evidence.md) records the +independent evidence. Azure Service Bus and Google Cloud Pub/Sub both distinguish competing +consumption on a queue from Topic fan-out through distinct Subscriptions, and from load balancing +inside one Subscription. OpenTelemetry keeps `messaging.destination.subscription.name` separate from +`messaging.consumer.group.name`. A Kafka consumer group is a partition/offset construct, not a +broker-held named Subscription. Neither positive broker needs a broker-specific production +exception. + +## Decision + +**`GO`.** AIP introduces the bounded source-independent Pub/Sub family defined by the I4 +specification. + +1. **Canonical family.** The only generic Pub/Sub entities are `Topic` and `Subscription`, and the + only relations are `Service -[PUBLISHES_TO]-> Topic`, `Subscription -[SUBSCRIPTION_OF]-> Topic`, + `Service -[RECEIVES_FROM]-> Subscription` and `Topic -[CARRIES]-> Message`. The existing Queue + model and Queue claim ids are unchanged. Fan-out is expressed only by distinct Subscriptions. +2. **Both ADR 0013 guards are retained.** `decide_destination_semantics` and + `decide_service_identity` stay in force, and I4 reuses them. It does not replace them, and it + does not add a second Service-identity implementation (I4 spec §9). +3. **No generic `Destination` normalization.** Queue, Topic and Subscription are type-distinct, + with distinct identity prefixes (`queue:`, `topic:`, `subscription:`). No supertype and no + name-based equivalence exists across them. +4. **No consumer-group-to-Subscription equivalence.** A consumer group may be bounded evidence + metadata. It never mints, resolves, or aliases a Subscription, and it is not an input to + Subscription identity. +5. **No runtime-only Topic/Subscription minting.** Runtime evidence may only qualify + already-declared Topic/Subscription topology. +6. **Bounded OpenTelemetry widening.** I4 cites ADR 0013 decision #2. Operation classification + continues to read only `messaging.operation.type`. Destination-side recognition widens by + exactly two keys, `messaging.destination.subscription.name` and `messaging.consumer.group.name`. + This is permitted because both guards named in decision #2 exist and remain in the path. +7. **No new public surface family.** I4 adds no MCP tool, keeping exactly three: no generic graph + tool, no live broker adapter, and no broker administration API. Topic/Subscription reach the + public contract only through the existing dependency, drift and evidence answers at + `schema_version` `0.5`. + +## Consequences + +- ADR 0013's Status line records that it is superseded in part. Its destination-semantics and + service-identity guards, and its "unsupported beats incorrectly supported" rule, remain + authoritative for everything outside the bounded family above. +- Topic kind requires positive evidence, and Subscription requires explicit stable identity. A + Channel, a `subscribe` operation, protocol/vendor, or `messaging.system` never establishes either. +- Subscription-specific dead-letter configuration differs materially across brokers. It is retained + only as an internal Subscription-scoped contribution (I4 spec §10), never as a guessed generic + target relation. +- The I4 stop conditions (spec §16) apply to any later change. In particular, a broker-specific + canonical entity or a broker-specific production exception returns this decision to review + instead of widening it silently. +- If I4 cannot complete, this ADR is superseded by a `DEFER` record rather than silently left + `Proposed`. diff --git a/docs/adr/README.md b/docs/adr/README.md index 5201d7a6..6be038ac 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -22,6 +22,7 @@ project's stated architecture principles. | [0014](0014-negotiated-mcp-client-interoperability.md) | Support negotiated MCP client interoperability without weakening the direct 2026-07-28 contract (Accepted) | | [0015](0015-bounded-reference-resolution.md) | Bounded multi-file `$ref` resolution is hand-rolled, not delegated to `referencing` (Accepted) | | [0016](0016-public-architecture-knowledge-adapters.md) | Use REST and standard negotiated MCP as the public Architecture Knowledge adapters (Proposed) | +| [0017](0017-source-independent-pubsub-semantics.md) | Source-independent Topic/Subscription semantics, with both ADR 0013 guards retained (Proposed; supersedes 0013 in part) | A new ADR is numbered sequentially and never renumbered or deleted — if a decision is superseded, add a new ADR and mark the old one's Status as `Superseded by NNNN`. @@ -32,4 +33,4 @@ and [0012](0012-observed-evidence-retention.md), when the benchmark and the rete it names are decided). 0009-0013 came out of [`architecture-review-0.4.0.md`](../architecture-review-0.4.0.md). 0014 came out of `v0.4.2` I1. 0015 came out of `v0.5.0` I1 PR3b. 0016 records the `v0.5.0` I3 public-adapter -consolidation decision. +consolidation decision. 0017 records the `v0.5.0` I4 Pub/Sub `GO` decision. diff --git a/docs/specifications/0.5.0/i4-decision-evidence.md b/docs/specifications/0.5.0/i4-decision-evidence.md new file mode 100644 index 00000000..17423bf5 --- /dev/null +++ b/docs/specifications/0.5.0/i4-decision-evidence.md @@ -0,0 +1,147 @@ +# AIP v0.5.0 I4 — Decision Evidence Record + +**Governing specification:** [`i4-source-independent-pubsub-semantics.md`](i4-source-independent-pubsub-semantics.md) +Draft 0.3, as merged at `f98e48b` (#226, #227 residuals)
+**Slice:** I4 Slice 1 — Decision and canonical foundation (spec §15)
+**Decision:** `GO` — recorded in [ADR 0017](../../adr/0017-source-independent-pubsub-semantics.md)
+**Evidence retrieval date:** 2026-09-23 + +This record is the §15 Slice 1 "independent evidence record". It is evidence, not qualification. +It shows that the §4 semantic distinctions are supported by independent public broker and +OpenTelemetry documentation. The deterministic broker-semantic fixtures in §13.3 remain slice 5 +work. + +--- + +## 1. Sources + +Quoted passages are short excerpts as retrieved on 2026-09-23. The URLs are the canonical public +documentation pages the spec cites in §4.1, plus two dead-letter pages cited for §10. + +| # | Source | URL | +|---|---|---| +| S1 | Azure Service Bus — Queues, topics, and subscriptions (`ms.date` 2026-01-31) | https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-queues-topics-subscriptions | +| S2 | Azure Service Bus — Dead-letter queues (`ms.date` 2026-07-16) | https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dead-letter-queues | +| S3 | Google Cloud Pub/Sub — Pub/Sub basics (was `cloud.google.com/pubsub/docs/pubsub-basics`, now 301 → `docs.cloud.google.com`) | https://docs.cloud.google.com/pubsub/docs/pubsub-basics | +| S4 | Google Cloud Pub/Sub — Handle message failures | https://docs.cloud.google.com/pubsub/docs/handling-failures | +| S5 | OpenTelemetry semantic conventions — Messaging attribute registry | https://opentelemetry.io/docs/specs/semconv/registry/attributes/messaging/ | +| S6 | OpenTelemetry semantic conventions — Messaging spans | https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/ | +| S7 | Apache Kafka 4.3 documentation — Design (the spec's `documentation/#design` anchor now resolves to a navigation hub; this is the current Design page) | https://kafka.apache.org/43/design/design/ | + +### Quoted evidence + +- **S1, Queues:** "Queues offer First In, First Out (FIFO) message delivery to one or more competing + consumers. ... only one message consumer receives and processes each message." +- **S1, Topics and subscriptions:** "Each published message is made available to each subscription + registered with the topic. Publisher sends a message to a topic and one or more subscribers + receive a copy of the message." Also: "consumers don't receive messages directly from the topic. + Instead, consumers receive messages from subscriptions of the topic. ... subscriptions support the + same patterns described earlier in this section regarding queues: competing consumer, temporal + decoupling, load leveling, and load balancing." +- **S2:** "Each queue and each subscription has its own dead-letter sub-queue." It is addressed as + `/Subscriptions//$deadletterqueue`. +- **S3, Fan-out:** "a single topic is attached to multiple subscriptions. ... Each of the subscriber + applications gets the same set of published messages from the topic." +- **S3, Load balancing:** "a single topic is attached to a single subscription that is, in turn, + connected to multiple subscriber applications. Each of the subscriber applications gets a subset + of the published messages, and no two subscriber applications get the same subset." +- **S4:** "You configure a dead-letter topic on a subscription, not on the topic it pulls from. + This is because it's a subscription property." +- **S5:** `messaging.destination.subscription.name` is "The name of the destination subscription + from which a message is consumed." `messaging.consumer.group.name` is "The name of the consumer + group with which a consumer is associated", and it carries the note that "Semantic conventions + for individual messaging systems SHOULD document whether `messaging.consumer.group.name` is + applicable and what it means in the context of that system." Both are *Development* stability, + and they are two distinct registry attributes. +- **S7:** "each partition is consumed by exactly one consumer within each subscribing consumer + group at any given time." The position of a consumer is tracked per partition as an offset. Group + semantics are therefore expressed through partition assignment and offsets, the constructs spec + §3.2/§6.2 exclude from the generic model. +- **S6:** `messaging.destination.name` "SHOULD uniquely identify a specific queue, topic or other + entity within the broker". No destination-kind attribute is defined in the current conventions. + +--- + +## 2. §4 distinctions → evidence + +| §4 distinction | Supported by | Disposition | +|---|---|---| +| `Queue != Topic` | S1: queue = single consumer per message; topic = copies per subscription | Supported | +| `Topic != Subscription` | S1: consumers receive from subscriptions, not the topic. S3: topic attached to subscriptions | Supported | +| `AsyncAPI Channel != broker destination` | Unchanged I1 §9 rule (Channel is a source-language construct). No broker source says otherwise | Retained | +| `consumer instance != Subscription` | S1: consumers compete within a subscription. S3: multiple subscriber applications share one subscription's messages | Supported | +| `consumer group != automatically Subscription` | S5: distinct attributes, with group meaning left to each system. S7: group semantics are partition/offset-based | Supported (Kafka negative boundary) | +| Queue + multiple consumers → competing consumers, no fan-out | S1 | Supported | +| Topic + multiple Subscriptions → fan-out | S1, S3 | Supported | +| one Subscription + multiple consumer instances → load balancing, not fan-out | S1, S3 | Supported | +| same destination name → no identity equivalence across kind/broker/namespace | S1 (queues and topics are separate entity kinds in one namespace). Spec §7 identity is broker/namespace scoped with type-distinct prefixes | Retained by identity formula | +| runtime destination name alone → insufficient kind evidence | S6: `messaging.destination.name` "SHOULD uniquely identify a specific queue, topic or other entity within the broker", and the current conventions define no generic destination-kind attribute (AIP's recognized `messaging.destination_kind` is a legacy key) | Supported | +| runtime consumer-group name alone → insufficient Subscription identity | S5 | Supported | + +## 3. §4.2 GO condition + +The abstraction in §4.2 is supported by two materially different positive brokers: + +- **Azure Service Bus (S1/S2):** a Subscription is a broker-held named entity under a Topic, and + consumers compete within it. **Google Cloud Pub/Sub (S3/S4):** a Subscription is a named entity + attached to one Topic, and subscribers load-balance within it. Both map to + `Topic ← SUBSCRIPTION_OF ← Subscription ← RECEIVES_FROM ← Service` with no product-specific + entity. +- **Kafka (S7)** has no broker-held named Subscription. A consumer group is a distinct + partition/offset construct, and is modeled only as the §4 negative boundary. + +**No broker-specific production exception is required** for either positive broker. The one place +the two differ materially is dead-letter handling: ASB has a per-subscription dead-letter subqueue +(S2), and Google has a dead-letter *topic* configured as a subscription property (S4). The spec +already handles this difference generically through §10's internal, Subscription-scoped +`SubscriptionDeadLetterConfiguration` carrier, which forces no target kind. That difference is the +reason the spec forbids a generic `Subscription -[DEAD_LETTERS_TO]-> Queue`. It is not an exception +to it. + +**Outcome: `GO`.** The §4.2 condition "If either positive broker fixture requires a broker-specific +production exception, the outcome SHALL be `DEFER`" does not trigger on this evidence. The slice 5 +fixtures (§13.3) remain the executable re-check. If a fixture later needs an exception, that is +stop condition §16 #11, and the decision returns to review. + +### Disclosed, out-of-scope constructs + +These remain unsupported and deferred, per §3.2: + +- ASB subscription filters/rules and actions (S1); +- ASB transfer dead-letter queues and auto-forwarding (S2); +- ASB JMS shared/unshared durable subscriptions and express entities (S1); +- Kafka partitions, offsets, and consumer rebalancing (S7); +- Kafka share groups (KIP-932; named here from general knowledge, not independently retrieved + for this record). + +None of these constructs becomes Subscription or Queue semantics in I4. + +--- + +## 4. §19 entry-gate checklist dispositions + +| Checklist item | Disposition | +|---|---| +| I4 decision is explicit GO or DEFER | **GO**, recorded here and in ADR 0017 (Status `Proposed` until the I4 work lands, per the ADR index convention) | +| Azure Service Bus and Google Pub/Sub independently support the abstraction | Satisfied, §3 above. Executable fixtures: slice 5 | +| Kafka consumer group is not normalized to Subscription | Satisfied by spec §4/§7.2/§9. Slice 1's `subscription_owned_id` has no consumer-group input (pinned by test). Runtime guard: slice 3. Fixture: slice 5 | +| Queue remains competing-consumer semantics | Satisfied. Queue model, identity and claim ids are unchanged in slice 1 | +| Queue and Subscription each preserve one resolved claim per distinct evidenced logical consumer without calling it fan-out | Specified in §6.3/§12.3. Projection: slice 4 | +| Topic fan-out is represented by distinct Subscriptions | Specified in §6.3. Slice 1 `Subscription` carries no consumer fields. Projection: slice 4 | +| Multiple instances on one Subscription are not fan-out | Specified in §6.3/§12.3. Slice 4/5 | +| AsyncAPI Channel is not automatically Queue/Topic | Specified in §8. Slice 2 | +| `x-aip-destination-kind` is bounded to `queue\|topic` | Specified in §8.1. Slice 2 | +| Topic identity requires stable broker/namespace evidence | Slice 1 `topic_owned_id` takes the broker id and namespace as required inputs. Evidence sourcing: slice 2 | +| Subscription identity includes Topic id | Slice 1 `subscription_owned_id` binds the canonical Topic id (pinned by test) | +| `subscribe` direction alone cannot mint Subscription | Specified in §8.3. Slice 2 | +| Runtime cannot mint Topic/Subscription | Specified in §9. Slice 3 | +| Consumer-group name cannot resolve Subscription | Specified in §9. Slice 3 | +| Existing v0.4.1 destination/service guards remain active | Unchanged in slice 1. ADR 0017 retains both | +| I3 `DEPLOYED_AS` and Kubernetes placement are not Pub/Sub evidence | Specified in §2. No slice 1 code reads them | +| Exactly three MCP tools remain | Unchanged in slice 1 (existing tool-count tests) | +| `ArchitectureIntelligenceService` remains semantic owner | Unchanged in slice 1 | +| Queue claim ids remain unchanged | Slice 1 leaves the claim-id payload unchanged. The optional `subscription_id` lands in slice 4 | +| Canonicalization version bumps to 3, with Topic/Subscription node queries, in the first slice that persists new public state | Slice 2. Slice 1 pins `_CANONICALIZATION_VERSION == 2` and a closed persistence path | +| Schema version remains `0.5` | Satisfied. The slice 1 schema widening stays at `0.5` | +| No live broker adapter is added | Satisfied | +| No broker-specific production branch is required | Satisfied on this evidence (§3). Slice 5 re-checks it executably | diff --git a/schemas/architecture_intelligence/v0.5/architecture-answer.schema.json b/schemas/architecture_intelligence/v0.5/architecture-answer.schema.json index 78edcb5b..3a581f2a 100644 --- a/schemas/architecture_intelligence/v0.5/architecture-answer.schema.json +++ b/schemas/architecture_intelligence/v0.5/architecture-answer.schema.json @@ -66,10 +66,33 @@ ] }, "then": { + "properties": { + "relation_type": { + "enum": [ + "PUBLISHES_TO", + "SENDS" + ] + } + }, + "required": [ + "relation_type", + "via" + ] + } + }, + { + "if": { "properties": { "relation_type": { "const": "SENDS" - }, + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { "via": { "properties": { "type": { @@ -82,10 +105,85 @@ } }, "required": [ - "relation_type", "via" ] } + }, + { + "if": { + "properties": { + "relation_type": { + "const": "PUBLISHES_TO" + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "else": { + "properties": { + "subscription": { + "type": "null" + } + } + }, + "if": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "properties": { + "subscription": { + "anyOf": [ + { + "type": "null" + }, + { + "properties": { + "type": { + "const": "SUBSCRIPTION" + } + }, + "required": [ + "type" + ] + } + ] + } + } } ], "properties": { @@ -95,6 +193,17 @@ "relation_type": { "$ref": "#/$defs/DeliveryRelationType" }, + "subscription": { + "anyOf": [ + { + "$ref": "#/$defs/EntityRef" + }, + { + "type": "null" + } + ], + "default": null + }, "via": { "$ref": "#/$defs/EntityRef" } @@ -110,7 +219,8 @@ "DeliveryRelationType": { "enum": [ "CALLS", - "SENDS" + "SENDS", + "PUBLISHES_TO" ], "title": "DeliveryRelationType", "type": "string" @@ -636,7 +746,11 @@ "if": { "properties": { "type": { - "const": "QUEUE" + "enum": [ + "QUEUE", + "SUBSCRIPTION", + "TOPIC" + ] } }, "required": [ @@ -719,7 +833,9 @@ "SERVICE", "OPERATION", "QUEUE", - "WORKLOAD" + "WORKLOAD", + "TOPIC", + "SUBSCRIPTION" ], "title": "EntityType", "type": "string" diff --git a/schemas/architecture_intelligence/v0.5/drift-answer.schema.json b/schemas/architecture_intelligence/v0.5/drift-answer.schema.json index ed73830c..c5313e46 100644 --- a/schemas/architecture_intelligence/v0.5/drift-answer.schema.json +++ b/schemas/architecture_intelligence/v0.5/drift-answer.schema.json @@ -101,10 +101,33 @@ ] }, "then": { + "properties": { + "relation_type": { + "enum": [ + "PUBLISHES_TO", + "SENDS" + ] + } + }, + "required": [ + "relation_type", + "via" + ] + } + }, + { + "if": { "properties": { "relation_type": { "const": "SENDS" - }, + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { "via": { "properties": { "type": { @@ -117,10 +140,85 @@ } }, "required": [ - "relation_type", "via" ] } + }, + { + "if": { + "properties": { + "relation_type": { + "const": "PUBLISHES_TO" + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "else": { + "properties": { + "subscription": { + "type": "null" + } + } + }, + "if": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "properties": { + "subscription": { + "anyOf": [ + { + "type": "null" + }, + { + "properties": { + "type": { + "const": "SUBSCRIPTION" + } + }, + "required": [ + "type" + ] + } + ] + } + } } ], "properties": { @@ -130,6 +228,17 @@ "relation_type": { "$ref": "#/$defs/DeliveryRelationType" }, + "subscription": { + "anyOf": [ + { + "$ref": "#/$defs/EntityRef" + }, + { + "type": "null" + } + ], + "default": null + }, "via": { "$ref": "#/$defs/EntityRef" } @@ -145,7 +254,8 @@ "DeliveryRelationType": { "enum": [ "CALLS", - "SENDS" + "SENDS", + "PUBLISHES_TO" ], "title": "DeliveryRelationType", "type": "string" @@ -502,7 +612,11 @@ "if": { "properties": { "type": { - "const": "QUEUE" + "enum": [ + "QUEUE", + "SUBSCRIPTION", + "TOPIC" + ] } }, "required": [ @@ -585,7 +699,9 @@ "SERVICE", "OPERATION", "QUEUE", - "WORKLOAD" + "WORKLOAD", + "TOPIC", + "SUBSCRIPTION" ], "title": "EntityType", "type": "string" diff --git a/schemas/architecture_intelligence/v0.5/evidence-answer.schema.json b/schemas/architecture_intelligence/v0.5/evidence-answer.schema.json index 4259c2a1..c28b0b0d 100644 --- a/schemas/architecture_intelligence/v0.5/evidence-answer.schema.json +++ b/schemas/architecture_intelligence/v0.5/evidence-answer.schema.json @@ -66,10 +66,33 @@ ] }, "then": { + "properties": { + "relation_type": { + "enum": [ + "PUBLISHES_TO", + "SENDS" + ] + } + }, + "required": [ + "relation_type", + "via" + ] + } + }, + { + "if": { "properties": { "relation_type": { "const": "SENDS" - }, + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { "via": { "properties": { "type": { @@ -82,10 +105,85 @@ } }, "required": [ - "relation_type", "via" ] } + }, + { + "if": { + "properties": { + "relation_type": { + "const": "PUBLISHES_TO" + } + }, + "required": [ + "relation_type" + ] + }, + "then": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "else": { + "properties": { + "subscription": { + "type": "null" + } + } + }, + "if": { + "properties": { + "via": { + "properties": { + "type": { + "const": "TOPIC" + } + }, + "required": [ + "type" + ] + } + }, + "required": [ + "via" + ] + } + }, + { + "properties": { + "subscription": { + "anyOf": [ + { + "type": "null" + }, + { + "properties": { + "type": { + "const": "SUBSCRIPTION" + } + }, + "required": [ + "type" + ] + } + ] + } + } } ], "properties": { @@ -95,6 +193,17 @@ "relation_type": { "$ref": "#/$defs/DeliveryRelationType" }, + "subscription": { + "anyOf": [ + { + "$ref": "#/$defs/EntityRef" + }, + { + "type": "null" + } + ], + "default": null + }, "via": { "$ref": "#/$defs/EntityRef" } @@ -110,7 +219,8 @@ "DeliveryRelationType": { "enum": [ "CALLS", - "SENDS" + "SENDS", + "PUBLISHES_TO" ], "title": "DeliveryRelationType", "type": "string" @@ -467,7 +577,11 @@ "if": { "properties": { "type": { - "const": "QUEUE" + "enum": [ + "QUEUE", + "SUBSCRIPTION", + "TOPIC" + ] } }, "required": [ @@ -550,7 +664,9 @@ "SERVICE", "OPERATION", "QUEUE", - "WORKLOAD" + "WORKLOAD", + "TOPIC", + "SUBSCRIPTION" ], "title": "EntityType", "type": "string" @@ -688,7 +804,7 @@ "type": "object" }, "EvidenceRelationType": { - "description": "v0.4.0 I2.1 - the 7 canonical graph relation kinds (spec \u00a711.2's `supports`). Deliberately its\nown closed enum rather than reusing `DeliveryRelationType` (only CALLS/SENDS) or a graph-layer\nstring - `get_evidence` describes existing facts, never a new architecture claim.", + "description": "v0.4.0 I2.1 - the closed set of canonical graph relation kinds (spec \u00a711.2's `supports`),\nwidened by I3 (DEPLOYED_AS) and v0.5.0 I4 (PUBLISHES_TO/SUBSCRIPTION_OF). Deliberately its own\nclosed enum rather than reusing `DeliveryRelationType` (only the delivery relations) or a\ngraph-layer string - `get_evidence` describes existing facts, never a new architecture claim.", "enum": [ "PROVIDES", "CALLS", @@ -697,7 +813,9 @@ "CARRIES", "CONFORMS_TO", "DEAD_LETTERS_TO", - "DEPLOYED_AS" + "DEPLOYED_AS", + "PUBLISHES_TO", + "SUBSCRIPTION_OF" ], "title": "EvidenceRelationType", "type": "string" diff --git a/tests/integration/test_importer.py b/tests/integration/test_importer.py index 3dad1ba5..86f7b2c7 100644 --- a/tests/integration/test_importer.py +++ b/tests/integration/test_importer.py @@ -111,6 +111,8 @@ def test_ensure_schema_creates_constraints(driver): "service_id", "operation_id", "queue_id", + "topic_id", + "subscription_id", "message_id", "schema_id", "source_state_source_instance_id", diff --git a/tests/unit/test_architecture_intelligence_contracts.py b/tests/unit/test_architecture_intelligence_contracts.py index 40a6854d..05acce2b 100644 --- a/tests/unit/test_architecture_intelligence_contracts.py +++ b/tests/unit/test_architecture_intelligence_contracts.py @@ -1824,3 +1824,145 @@ def test_drift_data_schema_requires_service_typed_service(): assert data_schema["allOf"][0]["properties"]["service"]["properties"]["type"]["const"] == ( "SERVICE" ) + + +# --- v0.5.0 I4 slice 1: Pub/Sub public contract skeleton (spec §12.1/§12.2) -------------------- + + +def _i4_ref(entity_type: EntityType) -> EntityRef: + if entity_type == EntityType.OPERATION: + return _valid_operation_entity() + return EntityRef(id=f"{entity_type.value.lower()}:x:y", type=entity_type, name="y") + + +def _i4_answer(claim: dict) -> dict: + # A complete, valid ServiceDependenciesData (incl. the I3 deployment fields) so each I4 + # parity case varies exactly one delivery/entity shape against an otherwise-valid answer. + return _answer_dict_with_claim_dict( + claim, + data={ + "service": _valid_service_entity().model_dump(mode="json"), + "dependency_claim_ids": [claim["claim_id"]], + "deployment_claim_ids": [], + "deployment_resolutions": [], + }, + ) + + +def _pydantic_and_schema_validity(payload: dict) -> tuple[bool, bool]: + try: + ANSWER_TYPE.model_validate(payload) + pydantic_valid = True + except ValidationError: + pydantic_valid = False + return pydantic_valid, jsonschema.Draft202012Validator(load_schema()).is_valid(payload) + + +def test_i4_parity_baseline_answer_is_valid_in_both(): + """Positive control: without it, every negative parity case below could pass vacuously.""" + assert _pydantic_and_schema_validity(_i4_answer(_valid_claim().model_dump(mode="json"))) == ( + True, + True, + ) + + +_I4_VIA_TYPES = [t for t in EntityType if t != EntityType.WORKLOAD] +_I4_SUBSCRIPTION_CHOICES = [None, EntityType.SUBSCRIPTION, EntityType.QUEUE, EntityType.TOPIC] +_I4_VALID_DELIVERIES = { + (DeliveryKind.SYNC_HTTP, DeliveryRelationType.CALLS, EntityType.OPERATION, None), + (DeliveryKind.ASYNC_MESSAGE, DeliveryRelationType.SENDS, EntityType.QUEUE, None), + (DeliveryKind.ASYNC_MESSAGE, DeliveryRelationType.PUBLISHES_TO, EntityType.TOPIC, None), + ( + DeliveryKind.ASYNC_MESSAGE, + DeliveryRelationType.PUBLISHES_TO, + EntityType.TOPIC, + EntityType.SUBSCRIPTION, + ), +} + + +@pytest.mark.parametrize("subscription_type", _I4_SUBSCRIPTION_CHOICES) +@pytest.mark.parametrize("via_type", _I4_VIA_TYPES) +@pytest.mark.parametrize("relation_type", list(DeliveryRelationType)) +@pytest.mark.parametrize("kind", list(DeliveryKind)) +def test_delivery_ref_cross_product_agrees_between_pydantic_and_frozen_schema( + kind, relation_type, via_type, subscription_type +): + """I4 spec §12.2: the exhaustive (kind, relation_type, via.type, subscription) cross product. + Exactly the four spec-listed shapes are valid, and Pydantic and the committed JSON Schema agree + on every combination (frozen contract <-> schema parity).""" + expected_valid = (kind, relation_type, via_type, subscription_type) in _I4_VALID_DELIVERIES + claim = _valid_claim().model_dump(mode="json") + claim["delivery"] = { + "kind": kind.value, + "relation_type": relation_type.value, + "via": _i4_ref(via_type).model_dump(mode="json"), + "subscription": ( + None + if subscription_type is None + else _i4_ref(subscription_type).model_dump(mode="json") + ), + } + pydantic_valid, schema_valid = _pydantic_and_schema_validity(_i4_answer(claim)) + + assert pydantic_valid is expected_valid + assert schema_valid is expected_valid + + +def test_delivery_ref_subscription_defaults_to_null_and_is_emitted(): + dumped = _valid_delivery().model_dump(mode="json") + assert "subscription" in dumped + assert dumped["subscription"] is None + + +@pytest.mark.parametrize("entity_type", _I4_VIA_TYPES) +def test_entity_ref_protocol_namespace_agree_between_pydantic_and_frozen_schema(entity_type): + """I4 spec §12.1: protocol/namespace are allowed for QUEUE/TOPIC/SUBSCRIPTION only.""" + expected_valid = entity_type in {EntityType.QUEUE, EntityType.TOPIC, EntityType.SUBSCRIPTION} + via = _i4_ref(entity_type).model_dump(mode="json") + via.update(protocol="amqp", namespace="ns") + claim = _valid_claim().model_dump(mode="json") + claim["object"] = via + claim["destination_resolution"] = DestinationResolution.DIRECT_TARGET_FALLBACK.value + claim["resolution_evidence_refs"] = [] + pydantic_valid, schema_valid = _pydantic_and_schema_validity(_i4_answer(claim)) + assert pydantic_valid is expected_valid + assert schema_valid is expected_valid + + +def test_i4_public_vocabulary_additions_are_exact(): + assert {EntityType.TOPIC.value, EntityType.SUBSCRIPTION.value} <= {t.value for t in EntityType} + assert {t.value for t in DeliveryRelationType} == {"CALLS", "SENDS", "PUBLISHES_TO"} + evidence_enum = load_evidence_schema()["$defs"]["EvidenceRelationType"]["enum"] + pre_i4 = { + "PROVIDES", + "CALLS", + "SENDS", + "RECEIVES_FROM", + "CARRIES", + "CONFORMS_TO", + "DEAD_LETTERS_TO", + "DEPLOYED_AS", + } + assert set(evidence_enum) - pre_i4 == {"PUBLISHES_TO", "SUBSCRIPTION_OF"} + assert pre_i4 <= set(evidence_enum) + + +def test_topic_and_subscription_claim_objects_are_admitted_as_direct_target_fallback(): + """I4 spec §12.3: Topic or Subscription may be a DIRECT_TARGET_FALLBACK claim object.""" + for object_type in (EntityType.TOPIC, EntityType.SUBSCRIPTION): + _valid_claim( + object=_i4_ref(object_type), + destination_resolution=DestinationResolution.DIRECT_TARGET_FALLBACK, + resolution_evidence_refs=[], + delivery=DeliveryRef( + kind=DeliveryKind.ASYNC_MESSAGE, + relation_type=DeliveryRelationType.PUBLISHES_TO, + via=_i4_ref(EntityType.TOPIC), + subscription=( + _i4_ref(EntityType.SUBSCRIPTION) + if object_type == EntityType.SUBSCRIPTION + else None + ), + ), + ) diff --git a/tests/unit/test_canonical_pubsub_foundation.py b/tests/unit/test_canonical_pubsub_foundation.py new file mode 100644 index 00000000..c88e0c5c --- /dev/null +++ b/tests/unit/test_canonical_pubsub_foundation.py @@ -0,0 +1,51 @@ +"""v0.5.0 I4 slice 1 (spec §6.2, §11, §15): the in-memory Topic/Subscription foundation, and the +fail-closed guard that no Pub/Sub node or relation can be persisted before slice 2. + +Slice 2 opens the persistence path atomically - `ArchitectureModel` Topic/Subscription lists, +importer `NODE_LABELS`/`KNOWN_RELATION_TYPES`, the `graph_schema` `RELATIONS` registry, dedicated +Topic/Subscription canonicalization queries, and the `_CANONICALIZATION_VERSION` 2 -> 3 bump - and +must update `test_pubsub_persistence_path_is_closed_until_slice_2` in that same commit. +""" + +from app.architecture_intelligence import repository +from app.canonical.model import ArchitectureModel, Subscription, Topic +from app.graph.importer import KNOWN_RELATION_TYPES, NODE_LABELS +from app.graph.schema import CONSTRAINTS +from app.graph_schema.registry import RELATIONS + + +def test_topic_and_subscription_carry_exactly_the_spec_fields(): + expected = {"id", "name", "protocol", "namespace"} + assert set(Topic.model_fields) == expected + assert set(Subscription.model_fields) == expected + + +def test_topic_and_subscription_optional_metadata_defaults_to_none(): + topic = Topic(id="topic:owned:" + "a" * 64, name="orders") + subscription = Subscription(id="subscription:owned:" + "b" * 64, name="billing") + assert (topic.protocol, topic.namespace) == (None, None) + assert (subscription.protocol, subscription.namespace) == (None, None) + + +def test_pubsub_persistence_path_is_closed_until_slice_2(): + """I4 spec §11: "Slice 1 may add in-memory models/schema skeletons only; it SHALL persist no + Pub/Sub node or relation." Persisting any Pub/Sub relation before the canonicalization-v3 bump + is prohibited, so every persistence/NL-query entry point still excludes Pub/Sub here.""" + model_fields = set(ArchitectureModel.model_fields) + assert not model_fields & {"topics", "subscriptions"} + assert not {"Topic", "Subscription"} & set(NODE_LABELS.values()) + assert not {"PUBLISHES_TO", "SUBSCRIPTION_OF"} & set(KNOWN_RELATION_TYPES) + assert not {"PUBLISHES_TO", "SUBSCRIPTION_OF"} & set(RELATIONS) + for relation in RELATIONS.values(): + assert not {"Topic", "Subscription"} & (relation.source_labels | relation.target_labels) + assert repository._CANONICALIZATION_VERSION == 2 + + +def test_topic_and_subscription_uniqueness_constraints_are_declared(): + assert "CREATE CONSTRAINT topic_id IF NOT EXISTS FOR (t:Topic) REQUIRE t.id IS UNIQUE" in ( + CONSTRAINTS + ) + assert ( + "CREATE CONSTRAINT subscription_id IF NOT EXISTS " + "FOR (s:Subscription) REQUIRE s.id IS UNIQUE" + ) in CONSTRAINTS diff --git a/tests/unit/test_sources_model.py b/tests/unit/test_sources_model.py index b528a277..bc118341 100644 --- a/tests/unit/test_sources_model.py +++ b/tests/unit/test_sources_model.py @@ -87,3 +87,21 @@ def test_diagnostic_code_includes_pr3b_reference_resolution_codes(): DiagnosticCode.REFERENCE_INVALID, DiagnosticCode.UNSUPPORTED_DIALECT_VERSION, } <= set(DiagnosticCode) + + +def test_i4_diagnostic_code_addition_is_exactly_the_three_spec_members(): + """v0.5.0 I4 spec §8.4/§13.1: I4 adds exactly TOPIC_IDENTITY_CONFLICT, + SUBSCRIPTION_IDENTITY_CONFLICT, and SUBSCRIPTION_IDENTITY_MISSING - and no + DESTINATION_KIND_UNSUPPORTED member.""" + i4_codes = { + "TOPIC_IDENTITY_CONFLICT", + "SUBSCRIPTION_IDENTITY_CONFLICT", + "SUBSCRIPTION_IDENTITY_MISSING", + } + members = {member.value for member in DiagnosticCode} + assert i4_codes <= members + assert "DESTINATION_KIND_UNSUPPORTED" not in members + new_topic_or_subscription_codes = { + value for value in members if value.startswith(("TOPIC_", "SUBSCRIPTION_")) + } + assert new_topic_or_subscription_codes == i4_codes diff --git a/tests/unit/test_sources_owner_ids.py b/tests/unit/test_sources_owner_ids.py index 1d6f80df..5fe0be0b 100644 --- a/tests/unit/test_sources_owner_ids.py +++ b/tests/unit/test_sources_owner_ids.py @@ -1,3 +1,6 @@ +import hashlib +import inspect + import pytest from app.sources.owner_ids import ( @@ -8,6 +11,8 @@ normalize_x_version, queue_owned_id, schema_owned_id, + subscription_owned_id, + topic_owned_id, ) @@ -158,3 +163,104 @@ def test_queue_owned_id_requires_broker_namespace_and_channel_agreement(): ) assert len({base, different_broker, different_namespace, different_channel}) == 4 assert base.startswith("queue:owned:") + + +# --- v0.5.0 I4 spec §7.1/§7.2 ------------------------------------------------------------------ + + +def _independent_length_delimited(*parts: str) -> bytes: + # Written from the I1 §5 encoding rule (8-byte big-endian length prefix per UTF-8 part), not by + # calling app.sources.encoding.length_delimited - an independent golden-vector derivation. + return b"".join(len(p.encode()).to_bytes(8, "big") + p.encode() for p in parts) + + +def test_topic_owned_id_matches_the_independently_derived_formula(): + expected = ( + "topic:owned:" + + hashlib.sha256( + _independent_length_delimited("broker:asb:commerce", "ns", "orders") + ).hexdigest() + ) + assert ( + topic_owned_id( + stable_broker_id="broker:asb:commerce", + normalized_namespace_or_empty="ns", + exact_topic_address="orders", + ) + == expected + ) + + +def test_subscription_owned_id_matches_the_independently_derived_formula(): + topic_id = "topic:owned:" + "a" * 64 + expected = ( + "subscription:owned:" + + hashlib.sha256( + _independent_length_delimited("broker:asb:commerce", "", topic_id, "billing") + ).hexdigest() + ) + assert ( + subscription_owned_id( + stable_broker_id="broker:asb:commerce", + normalized_namespace_or_empty="", + topic_id=topic_id, + exact_subscription_name="billing", + ) + == expected + ) + + +def test_topic_owned_id_requires_broker_namespace_and_address_agreement(): + def topic(broker="broker:asb:commerce", namespace="", address="orders"): + return topic_owned_id( + stable_broker_id=broker, + normalized_namespace_or_empty=namespace, + exact_topic_address=address, + ) + + ids = {topic(), topic(broker="broker:asb:other"), topic(namespace="ns"), topic(address="x")} + assert len(ids) == 4 + + +def test_queue_and_topic_ids_never_alias_for_identical_inputs(): + """I4 spec §7.2: Queue, Topic, and Subscription ids use distinct prefixes and never alias merely + because names match.""" + inputs = {"stable_broker_id": "b", "normalized_namespace_or_empty": "n"} + queue = queue_owned_id(**inputs, exact_channel_address="orders") + topic = topic_owned_id(**inputs, exact_topic_address="orders") + subscription = subscription_owned_id(**inputs, topic_id=topic, exact_subscription_name="orders") + assert queue.startswith("queue:owned:") + assert topic.startswith("topic:owned:") + assert subscription.startswith("subscription:owned:") + # Queue and Topic share the owner-key formula; only the type prefix separates them. + assert queue.removeprefix("queue:owned:") == topic.removeprefix("topic:owned:") + assert len({queue, topic, subscription}) == 3 + + +def test_same_subscription_name_under_different_topics_is_distinct(): + def subscription(topic_id): + return subscription_owned_id( + stable_broker_id="b", + normalized_namespace_or_empty="", + topic_id=topic_id, + exact_subscription_name="billing", + ) + + orders = topic_owned_id( + stable_broker_id="b", normalized_namespace_or_empty="", exact_topic_address="orders" + ) + invoices = topic_owned_id( + stable_broker_id="b", normalized_namespace_or_empty="", exact_topic_address="invoices" + ) + assert subscription(orders) != subscription(invoices) + + +def test_subscription_owned_id_has_no_consumer_group_input(): + """I4 spec §7.2: "A consumer-group identifier SHALL NOT be fed into the Subscription identity + formula." Pinned structurally on the helper's signature.""" + assert list(inspect.signature(subscription_owned_id).parameters) == [ + "stable_broker_id", + "normalized_namespace_or_empty", + "topic_id", + "exact_subscription_name", + ] From eaa2ae85a52e4e5c417ce0cf0425fd8d1d12a49e Mon Sep 17 00:00:00 2001 From: Michael Egner Date: Wed, 23 Sep 2026 20:40:14 +0200 Subject: [PATCH 2/2] test: pin keyword-only signature of Pub/Sub owned-id helpers Addresses PR #228 review: the signature test pinned parameter names but not the keyword-only contract shared with queue_owned_id. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_sources_owner_ids.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/unit/test_sources_owner_ids.py b/tests/unit/test_sources_owner_ids.py index 5fe0be0b..b7f2c03a 100644 --- a/tests/unit/test_sources_owner_ids.py +++ b/tests/unit/test_sources_owner_ids.py @@ -264,3 +264,13 @@ def test_subscription_owned_id_has_no_consumer_group_input(): "topic_id", "exact_subscription_name", ] + + +@pytest.mark.parametrize("helper", [topic_owned_id, subscription_owned_id]) +def test_pubsub_owned_id_helpers_are_keyword_only(helper): + """Mirrors `queue_owned_id`: every identity input is keyword-only, so two same-typed string + inputs (e.g. broker id and namespace) can never be transposed positionally.""" + parameters = inspect.signature(helper).parameters.values() + assert all(p.kind is inspect.Parameter.KEYWORD_ONLY for p in parameters) + with pytest.raises(TypeError): + helper(*(["x"] * len(inspect.signature(helper).parameters)))