feat: bind egress to exact package hops

adapters/http/app.py

@@ -26,10 +26,12 @@
 from adapters.http.contracts import (
     AcquireWire,
     AuthenticationFailureWire,
+    ChannelEgressGrantWire,
     CitationNotAvailableWire,
     ContextPackageWire,
     ContinueWire,
     InvalidRequestWire,
+    ModelEgressGrantWire,
     OpenCitationWire,
     RequestNotAvailableWire,
     ResolutionOutcomeWire,
@@ -98,6 +100,11 @@
     private_delivery_audience_digest_for_binding,
     redeem_private_delivery_evidence,
 )
+from engine.runtime.egress import (
+    ChannelEgressGrant,
+    EgressGrantIssuanceUnavailable,
+    ModelEgressGrant,
+)
 from engine.runtime.invocation import (
     _construct_authenticated_http_invocation,
 )
@@ -606,6 +613,8 @@ def resolve_context(
             raise TrustedAuthorityUnavailable from None
         except ContextRunPersistenceUnavailable:
             raise TrustedAuthorityUnavailable from None
+        except EgressGrantIssuanceUnavailable:
+            raise TrustedAuthorityUnavailable from None
         except ScopeAuthorityUnavailable:
             raise TrustedAuthorityUnavailable from None
         except InvalidTrustedScopeSnapshot:
@@ -679,11 +688,23 @@ def _resolution_outcome_to_wire(
 
 
 def _resolved_to_wire(outcome: Resolved) -> ResolvedWire:
+    egress_grant: ModelEgressGrantWire | ChannelEgressGrantWire | None = None
+    if type(outcome.egress_grant) is ModelEgressGrant:
+        egress_grant = ModelEgressGrantWire(
+            kind="model",
+            value=outcome.egress_grant.value,
+        )
+    elif type(outcome.egress_grant) is ChannelEgressGrant:
+        egress_grant = ChannelEgressGrantWire(
+            kind="channel",
+            value=outcome.egress_grant.value,
+        )
     return ResolvedWire(
         kind=outcome.kind,
         package=ContextPackageWire.model_validate(
             context_package_public_document(outcome.package)
         ),
+        egressGrant=egress_grant,
     )
 
 

adapters/http/contracts.py

@@ -65,6 +65,14 @@
     str,
     Field(strict=True, pattern=r"^[0-9a-f]{64}$"),
 ]
+OpaqueModelEgressGrant = Annotated[
+    str,
+    Field(strict=True, pattern=r"^egrm_[0-9a-f]{64}$", repr=False),
+]
+OpaqueChannelEgressGrant = Annotated[
+    str,
+    Field(strict=True, pattern=r"^egrc_[0-9a-f]{64}$", repr=False),
+]
 BlockOutputRef = Annotated[
     str,
     Field(strict=True, pattern=r"^block_[0-9a-f]{64}$"),
@@ -313,11 +321,30 @@ def require_exact_authorized_content_closure(self) -> Self:
         return self
 
 
+class ModelEgressGrantWire(ClosedWireModel):
+    """Opaque one-hop model grant; no trusted claim is exposed on the wire."""
+
+    kind: Literal["model"]
+    value: OpaqueModelEgressGrant = Field(repr=False)
+
+
+class ChannelEgressGrantWire(ClosedWireModel):
+    """Opaque one-hop channel grant; it carries no write authority."""
+
+    kind: Literal["channel"]
+    value: OpaqueChannelEgressGrant = Field(repr=False)
+
+
 class ResolvedWire(ClosedWireModel):
     """Successful public resolution envelope."""
 
     kind: Literal["resolved"]
     package: ContextPackageWire
+    egressGrant: ModelEgressGrantWire | ChannelEgressGrantWire | None = Field(
+        default=None,
+        discriminator="kind",
+        repr=False,
+    )
 
 
 class RequestNotAvailableWire(ClosedWireModel):

bot_delivery/__init__.py

@@ -0,0 +1 @@
+"""Trusted BotDelivery module seams; the application process remains inactive."""