v0.5.0 I4 slice 1: decision and canonical foundation

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
 
 

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

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",

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):

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:<sha256(topic_owner_key)>
+
+    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:<sha256(subscription_owner_key)>
+
+    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)}"

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
 

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`.

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.