Add new event queues and use Queue enum throughout codebase

netengine/api/routes.py

@@ -196,6 +196,34 @@ async def teardown_world(
 # ─────────────────────────────────────────────
 
 
+class ServiceToggleRequest(BaseModel):
+    enabled: bool
+
+
+@router.put("/services/{name}")
+async def update_service(
+    name: str, body: ServiceToggleRequest, user: dict = Depends(require_auth)
+) -> dict[str, Any]:
+    """Enable or disable a named world service (mail, storage) in the spec."""
+    state = RuntimeState.load()
+
+    if not state.world_spec:
+        raise HTTPException(status_code=409, detail="No world spec loaded")
+
+    world_services = state.world_spec.get("world_services") or {}
+    if name not in world_services:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Service '{name}' not found — known services: {list(world_services)}",
+        )
+
+    world_services[name]["enabled"] = body.enabled
+    state.world_spec["world_services"] = world_services
+    state.save()
+
+    return {"status": "updated", "service": name, "enabled": body.enabled}
+
+
 @router.get("/services")
 async def get_services(user: dict = Depends(require_auth)) -> dict[str, Any]:
     """List running NetEngines containers and their status."""
@@ -404,6 +432,44 @@ async def create_and(body: ANDCreateRequest, user: dict = Depends(require_auth))
     return {"status": "provisioned", "and": body.name, "org": body.org, "dns_suffix": dns_suffix}
 
 
+class ANDProfileUpdateRequest(BaseModel):
+    profile: str
+    dns_suffix: str = ""
+
+
+@router.put("/ands/{and_name}/profile")
+async def update_and_profile(
+    and_name: str, body: ANDProfileUpdateRequest, user: dict = Depends(require_auth)
+) -> dict[str, Any]:
+    """Update the profile (and optionally dns_suffix) of a provisioned AND instance."""
+    state = RuntimeState.load()
+
+    if not state.world_spec:
+        raise HTTPException(status_code=409, detail="No world spec loaded")
+
+    ands_out = state.ands_output or {}
+    instances: list[dict[str, Any]] = ands_out.get("instances", [])
+    instance = next((i for i in instances if i.get("and_name") == and_name), None)
+    if instance is None:
+        raise HTTPException(status_code=404, detail=f"AND instance '{and_name}' not found")
+    profiles = (state.world_spec.get("ands") or {}).get("profiles", {})
+    if body.profile not in profiles:
+        raise HTTPException(
+            status_code=422,
+            detail=f"Profile '{body.profile}' not defined in spec"
+            f" — known profiles: {list(profiles)}",
+        )
+
+    instance["profile"] = body.profile
+    if body.dns_suffix:
+        instance["dns_suffix"] = body.dns_suffix
+
+    state.ands_output = ands_out
+    state.save()
+
+    return {"status": "updated", "and": and_name, "profile": body.profile}
+
+
 @router.delete("/ands/{and_name}")
 async def remove_and(and_name: str, user: dict = Depends(require_auth)) -> dict[str, Any]:
     """Remove an AND instance."""
@@ -512,6 +578,71 @@ async def dns_query(
         raise HTTPException(status_code=503, detail=f"DNS query failed: {exc}")
 
 
+# ─────────────────────────────────────────────
+# Gateway
+# ─────────────────────────────────────────────
+
+
+class GatewayUpdateRequest(BaseModel):
+    real_internet_mode: str = ""
+    upstream_resolver_enabled: bool | None = None
+    upstream_resolver_ip: str = ""
+    cross_world_mode: str = ""
+
+
+@router.put("/gateway")
+async def update_gateway(
+    body: GatewayUpdateRequest, user: dict = Depends(require_auth)
+) -> dict[str, Any]:
+    """Update gateway portal configuration in the running spec.
+
+    Fields left at their zero-values are not modified. Changes take effect on
+    the next bootstrap cycle; live gateway reconfiguration requires a full reload.
+    """
+    state = RuntimeState.load()
+
+    if not state.world_spec:
+        raise HTTPException(status_code=409, detail="No world spec loaded")
+
+    gw = state.world_spec.get("gateway_portal") or {}
+    real_internet = gw.get("real_internet") or {}
+    cross_world = gw.get("cross_world") or {}
+
+    valid_ri_modes = {"isolated", "shadowed", "mirrored", "exposed", "custom"}
+    valid_cw_modes = {"none", "peered", "federated"}
+
+    if body.real_internet_mode:
+        if body.real_internet_mode not in valid_ri_modes:
+            raise HTTPException(
+                status_code=422,
+                detail=f"Invalid real_internet_mode '{body.real_internet_mode}'"
+                f" — valid: {valid_ri_modes}",
+            )
+        real_internet["mode"] = body.real_internet_mode
+
+    if body.upstream_resolver_enabled is not None:
+        real_internet["upstream_resolver_enabled"] = body.upstream_resolver_enabled
+
+    if body.upstream_resolver_ip:
+        real_internet["upstream_resolver_ip"] = body.upstream_resolver_ip
+
+    if body.cross_world_mode:
+        if body.cross_world_mode not in valid_cw_modes:
+            raise HTTPException(
+                status_code=422,
+                detail=f"Invalid cross_world_mode '{body.cross_world_mode}'"
+                f" — valid: {valid_cw_modes}",
+            )
+        cross_world["mode"] = body.cross_world_mode
+
+    gw["real_internet"] = real_internet
+    gw["cross_world"] = cross_world
+    state.world_spec["gateway_portal"] = gw
+    state.save()
+
+    return {"status": "updated", "gateway_portal": gw}
+
+
 # ─────────────────────────────────────────────
 # PKI
 # ─────────────────────────────────────────────
@@ -540,6 +671,50 @@ async def list_certs(user: dict = Depends(require_auth)) -> dict[str, Any]:
     }
 
 
+class PKIRotationPolicyUpdateRequest(BaseModel):
+    enabled: bool | None = None
+    default_interval_hours: int | None = None
+    default_warning_days: int | None = None
+    cert_type_overrides: dict[str, Any] | None = None
+
+
+@router.put("/pki/rotation-policy")
+async def update_pki_rotation_policy(
+    body: PKIRotationPolicyUpdateRequest, user: dict = Depends(require_auth)
+) -> dict[str, Any]:
+    """Update PKI certificate rotation policy in the running spec.
+
+    Only fields provided (non-None) are updated; omitted fields are left as-is.
+    Changes are picked up by the rotation worker on its next iteration without restart.
+    """
+    state = RuntimeState.load()
+
+    if not state.world_spec:
+        raise HTTPException(status_code=409, detail="No world spec loaded")
+
+    pki = state.world_spec.get("pki") or {}
+    policy = pki.get("rotation_policy") or {}
+
+    if body.enabled is not None:
+        policy["enabled"] = body.enabled
+    if body.default_interval_hours is not None:
+        if body.default_interval_hours < 1:
+            raise HTTPException(status_code=422, detail="default_interval_hours must be >= 1")
+        policy["default_interval_hours"] = body.default_interval_hours
+    if body.default_warning_days is not None:
+        if body.default_warning_days < 1:
+            raise HTTPException(status_code=422, detail="default_warning_days must be >= 1")
+        policy["default_warning_days"] = body.default_warning_days
+    if body.cert_type_overrides is not None:
+        policy["cert_type_overrides"] = body.cert_type_overrides
+
+    pki["rotation_policy"] = policy
+    state.world_spec["pki"] = pki
+    state.save()
+
+    return {"status": "updated", "rotation_policy": policy}
+
+
 # ─────────────────────────────────────────────
 # Identity
 # ─────────────────────────────────────────────

netengine/core/drift_controller.py

@@ -12,6 +12,7 @@
 from typing import Any, Optional
 
 from netengine.core.orchestrator import Orchestrator
+from netengine.events.queues import Queue
 from netengine.events.schema import EventEnvelope
 from netengine.handlers._base import BasePhaseHandler
 
@@ -330,7 +331,7 @@ async def _emit_drift_event(
                 emitted_by="drift_controller",
                 payload=payload,
             )
-            await self.orchestrator.context.pgmq_client.send("drift_events", event)
+            await self.orchestrator.context.pgmq_client.send(Queue.DRIFT_EVENTS, event)
             logger.debug(f"Drift event emitted: {event_type}")
         except Exception as e:
             logger.error(f"Failed to emit drift event: {e}")

netengine/core/phase_graph.py

@@ -43,4 +43,5 @@
     6: ["world_registry_output", "domain_registry_output"],
     7: ["identity_inworld_output"],
     8: ["ands_output"],
+    9: ["world_services_output"],
 }

netengine/events/queues.py

@@ -14,13 +14,21 @@ class Queue(StrEnum):
     AND_PROVISIONING = "and_provisioning"
     INWORLD_ADMISSIONS = "inworld_admissions"
     SERVICES_ADMISSIONS = "services_admissions"
+    AND_ADMISSIONS = "and_admissions"
+    PKI_CERT_ROTATION_EVENTS = "pki_cert_rotation_events"
+    DRIFT_EVENTS = "drift_events"
+    WORLD_HEALTH = "world_health"
 
     # Dead-letter queues (derived from primary names)
     DNS_UPDATES_DLQ = "dns_updates_dlq"
     OIDC_PROVISIONING_DLQ = "oidc_provisioning_dlq"
     AND_PROVISIONING_DLQ = "and_provisioning_dlq"
     INWORLD_ADMISSIONS_DLQ = "inworld_admissions_dlq"
     SERVICES_ADMISSIONS_DLQ = "services_admissions_dlq"
+    AND_ADMISSIONS_DLQ = "and_admissions_dlq"
+    PKI_CERT_ROTATION_EVENTS_DLQ = "pki_cert_rotation_events_dlq"
+    DRIFT_EVENTS_DLQ = "drift_events_dlq"
+    WORLD_HEALTH_DLQ = "world_health_dlq"
 
 
 # Primary queues only — used for metrics/introspection endpoints
@@ -30,4 +38,8 @@ class Queue(StrEnum):
     Queue.AND_PROVISIONING,
     Queue.INWORLD_ADMISSIONS,
     Queue.SERVICES_ADMISSIONS,
+    Queue.AND_ADMISSIONS,
+    Queue.PKI_CERT_ROTATION_EVENTS,
+    Queue.DRIFT_EVENTS,
+    Queue.WORLD_HEALTH,
 )

netengine/handlers/mail_handler.py

@@ -200,11 +200,11 @@ async def _inject_dns_records(self) -> list[str]:
         orgs_configured = []
 
         # Get all orgs from world registry
-        if not hasattr(spec, "world_registry") or not spec.world_registry.initial_orgs:
+        if not hasattr(spec, "world_registry") or not spec.world_registry.organizations:
             self.logger.warning("No orgs found in spec.world_registry")
             return orgs_configured
 
-        for org_spec in spec.world_registry.initial_orgs:
+        for org_spec in spec.world_registry.organizations:
             org_name = org_spec.name
             org_domain = f"{org_name}.internal"
 

netengine/monitoring/service.py

@@ -6,6 +6,7 @@
 
 from netengine.core.pgmq_client import PGMQClient
 from netengine.diagnostic import ProbeResult, ProbeStatus, build_runner
+from netengine.events.queues import Queue
 from netengine.events.schema import EventEnvelope
 from netengine.spec.models import NetEngineSpec
 
@@ -87,7 +88,7 @@ async def _run_probe_cycle(self) -> None:
         )
 
         try:
-            msg_id = await self._pgmq.send("world_health", event)
+            msg_id = await self._pgmq.send(Queue.WORLD_HEALTH, event)
             logger.debug(f"Published world_health event (msg_id: {msg_id})")
         except Exception as e:
             logger.error(f"Failed to publish world_health event: {e}", exc_info=True)

netengine/phases/phase_ands.py

@@ -15,6 +15,7 @@
 from datetime import UTC, datetime
 from typing import Any, Optional
 
+from netengine.events.queues import Queue
 from netengine.events.schema import EventEnvelope
 from netengine.handlers._base import BasePhaseHandler
 from netengine.handlers.context import PhaseContext
@@ -346,7 +347,7 @@ async def _consume_org_admission_events(
 
         while True:
             try:
-                msg = await context.pgmq_client.receive("and_admissions")
+                msg = await context.pgmq_client.receive(Queue.AND_ADMISSIONS)
                 if not msg:
                     await asyncio.sleep(1)
                     continue
@@ -355,7 +356,7 @@ async def _consume_org_admission_events(
                     envelope = EventEnvelope(**json.loads(msg["message"]))
 
                     if envelope.event_type != "org.admitted":
-                        await context.pgmq_client.delete("and_admissions", msg["msg_id"])
+                        await context.pgmq_client.delete(Queue.AND_ADMISSIONS, msg["msg_id"])
                         continue
 
                     payload = envelope.payload
@@ -364,7 +365,7 @@ async def _consume_org_admission_events(
 
                     if not org_name:
                         logger.warning("org.admitted event missing org_name")
-                        await context.pgmq_client.delete("and_admissions", msg["msg_id"])
+                        await context.pgmq_client.delete(Queue.AND_ADMISSIONS, msg["msg_id"])
                         continue
 
                     logger.info(f"Auto-provisioning AND for org: {org_name}")
@@ -379,13 +380,13 @@ def __init__(self, org: str, profile: str) -> None:
                     and_instance = SyntheticAND(org_name, and_profile)
                     await self._provision_and(context, docker, gateway, and_instance, ands_spec)
                     logger.info(f"Auto-provisioned AND for org {org_name}")
-                    await context.pgmq_client.delete("and_admissions", msg["msg_id"])
+                    await context.pgmq_client.delete(Queue.AND_ADMISSIONS, msg["msg_id"])
 
                 except Exception as e:
                     logger.error(f"Failed to process org admission event: {e}")
                     try:
                         await context.pgmq_client.archive_to_dlq(
-                            "and_admissions", msg["msg_id"], str(e)
+                            Queue.AND_ADMISSIONS, msg["msg_id"], str(e)
                         )
                     except Exception as dlq_err:
                         logger.error(f"Failed to archive to DLQ: {dlq_err}")