refactor(events): consolidate outbox-relay Kafka wiring into @fuzefront/core

.claude/agents/backend-engineer.md

@@ -4,7 +4,7 @@ model: sonnet
 description: Implements ONLY the backend slice of a feature — HTTP API/services, business logic, DB schema/migrations, events, and the backend's own unit tests — against a frozen API contract. Does NOT build UI, the independent test suite, deploy wiring, or docs. Use for backend implementation in a contract-first fan-out.
 # Figma is reserved for frontend-engineer; pure-code agent gets core tools only (no MCP).
 tools: Task, Bash, Glob, Grep, LS, Read, Edit, MultiEdit, Write, NotebookEdit, WebFetch, WebSearch, TodoWrite
-skills: [api-contract-first, feature-flags, verification-protocol, model-cascade]
+skills: [api-contract-first, feature-flags, logging, verification-protocol, model-cascade]
 ---
 
 You are a **backend engineer** for FuzeFront. You implement the **backend slice only**.

.claude/agents/frontend-engineer.md

@@ -5,7 +5,7 @@ description: Implements ONLY the UI slice of a feature — a design-system-first
 # SOLE owner of the Figma MCP plugin (design-to-code). All other domain agents have
 # Figma removed from their tool grant — it is reserved here for the UI/design-system slice.
 tools: "*"
-skills: [fuzefront-ui-package, design-system-inheritance, design-system-conformance, ui-frame-contract, frontend-design, feature-flags, ui-runtime-validation, verification-protocol, model-cascade]
+skills: [fuzefront-ui-package, design-system-inheritance, design-system-conformance, ui-frame-contract, frontend-design, feature-flags, logging, ui-runtime-validation, verification-protocol, model-cascade]
 ---
 
 You are a **frontend engineer**. You implement the **UI slice only**.

.claude/agents/fuzeinfra-expert.md

@@ -48,6 +48,13 @@ Apps attach via `networks: { FuzeInfra: { external: true } }` and reach services
 ## CI/CD (`.github/workflows/`)
 `deploy-prod.yml` (validate → ArgoCD sync), `helm-validate.yml` (helm lint + **kubeconform `-ignore-missing-schemas`** for Traefik CRDs), `infrastructure-tests.yml` (pytest against a live stack), `deploy-ec2.yml`, `claude-ci-autofix.yml` + `grafana-crit-fix.yml` (Claude-driven autofix bots), `auto-merge.yml`, `telegram-pr-merged.yml`, `update-ignore-list.yml`.
 
+## Read access to prod for anyone (including consuming repos) — `cluster-query`
+`.github/workflows/cluster-query.yml` (`workflow_dispatch`, `runs-on: staging`) is **self-service read-only `kubectl` against the prod cluster**, and it is **not** FuzeInfra-only — any repo whose token can write Actions on FuzeInfra can dispatch it. Tell consumers this rather than relaying cluster state by hand:
+```bash
+gh workflow run cluster-query.yml --repo izzywdev/FuzeInfra -f kubectl_args='-n <ns> get pods -o wide'
+```
+Guard (executable in `tests/test_cluster_query_guard.py`): a read verb must be present (`get describe logs top events version api-resources api-versions explain cluster-info config`); every mutating/exec token is refused; **`Secret` reads and `--raw` are blocked** because FuzeInfra's job logs are **public** and a read whose *output* is a credential leaks it (this happened on 2026-07-29 with `LITELLM_MASTER_KEY`; `--raw` would have printed the runner's cluster-admin kubeconfig). SealedSecrets are readable on purpose. The dispatch credential needs **Actions: write** — `FUZEINFRA_DISPATCH_TOKEN` (Contents-only) does not cover it. Consumer-facing doc: `docs/consuming-repos/CLUSTER_QUERY.md`. Recovering a live Secret value goes through the operator SSH path in `docs/SECRETS_MANAGEMENT.md` §4, never here.
+
 ## Gotchas (learned the hard way — verify they're still in the code)
 - **Prod is GitOps. Never hand-deploy or `kubectl patch`/`edit` prod resources** — ArgoCD `selfHeal` reverts out-of-band changes within seconds. Change `helm/fuzeinfra` (or values), commit to `main`, let ArgoCD sync. (This bit the Grafana dashboard fix: the kubectl patch didn't persist; it had to go through Git.)
 - **Grafana v13 table panels**: pre-v39 schemas fail with "Error loading: table". Fix = migrate `custom.displayMode` → `custom.cellOptions` and bump `schemaVersion` to `39` in the dashboard JSON, then ship via Git→ArgoCD. Note (as of last check): only `cluster-overview`, `fuzeinfra-services`, and `kubernetes-pods` are at v39 — `kubernetes-nodes`/`logs-explorer` are still 38 and `infrastructure-overview` is 27, so re-check before assuming a given dashboard is migrated. (Leave legitimate `legend.displayMode`/bargauge `displayMode` alone — only table-cell `custom.displayMode` needs the swap.)

.github/workflows/ci.yml

@@ -154,9 +154,12 @@ jobs:
       # core + shared/kafka must be built first — the jest moduleNameMapper
       # resolves them from their dist/ output.
       - name: build core + shared (jest deps)
+        # shared MUST build before core — @fuzefront/core now imports
+        # @fuzefront/shared/kafka (the schema registry), and tsc resolves it
+        # through shared/dist, so shared has to be current first.
         run: |
-          npm run -w @fuzefront/core build
           npm run -w @fuzefront/shared build
+          npm run -w @fuzefront/core build
 
       - name: security — API-token jest suite (DB mocked, no Postgres)
         run: |

agent-templates/schema/role-manifest.schema.json

@@ -80,6 +80,40 @@
     "metadata": {
       "type": "object",
       "description": "Passed through as agent `metadata` (free-form tracking)."
+    },
+    "a2a": {
+      "type": "object",
+      "additionalProperties": false,
+      "description": "OPTIONAL A2A discoverability/publication block. Mirrors the frozen contract FuzeAgent/agent-templates/contracts/a2a/v1/schema/role-a2a-extension.schema.json. Every field has a derived default, so no existing role.json needs it. The card projection reads role/name/description/services/metadata/coordinator regardless; this block only lets a role improve discoverability (examples/tags) or opt out of publication.",
+      "properties": {
+        "publish": {
+          "type": "boolean",
+          "default": true,
+          "description": "false hides this role from the public card. Still reachable on the EXTENDED card if the caller is allowlisted (authz.md §5)."
+        },
+        "extendedOnly": {
+          "type": "boolean",
+          "default": false,
+          "description": "true publishes this skill ONLY on the authenticated extended card, never on the anonymous /.well-known/agent-card.json. Use for skills whose mere existence is sensitive."
+        },
+        "tags": {
+          "type": "array",
+          "items": { "type": "string" },
+          "description": "Extra tags merged with the derived tags. Derived tags are never removed."
+        },
+        "examples": {
+          "type": "array",
+          "items": { "type": "string" },
+          "description": "Example prompts a caller can send to this skill — the primary signal a calling agent uses to decide fit. Absent examples make a skill effectively undiscoverable."
+        },
+        "inputModes": { "type": "array", "items": { "type": "string" } },
+        "outputModes": { "type": "array", "items": { "type": "string" } },
+        "scopes": {
+          "type": "array",
+          "items": { "type": "string" },
+          "description": "OAuth scopes required for THIS skill, projected into the skill's securityRequirements."
+        }
+      }
     }
   }
 }

agent-templates/sync/role_loader.py

@@ -4,9 +4,12 @@
 followed by the base guardrail block and any role-specific `system_append`.
 """
 import json
+import logging
 import os
 import re
 
+log = logging.getLogger(__name__)
+
 HERE = os.path.dirname(os.path.abspath(__file__))
 TEMPLATES_ROOT = os.path.dirname(HERE)                # agent-templates/
 REPO_ROOT = os.path.dirname(TEMPLATES_ROOT)           # repo root (personas live under .claude/agents)
@@ -76,14 +79,29 @@ def agent_payload(manifest):
     # or set-but-empty -> "") — the API rejects an empty/invalid url. Also drop the
     # matching mcp_toolset so the agent creates cleanly with only its configured servers;
     # re-provision after setting the URL to add the server + tool back.
+    #
+    # A drop is NEVER silent (FA-13): a required server logs at WARNING (its tools are
+    # missing until the URL is set), an `optional: true` server logs at INFO. A silent
+    # strip was the worst failure mode — an agent came up tool-less with no signal at all.
     servers = expand_env(manifest.get("mcp_servers", []))
     valid, dropped = [], set()
     for s in servers:
+        # `optional` is our own hint, never part of the API payload — strip it either way.
+        optional = bool(s.pop("optional", False))
         url = s.get("url", "")
         if url and "${" not in url:
             valid.append(s)
+            continue
+        name = s.get("name")
+        dropped.add(name)
+        reason = "url unset/empty" if not url else f"url unresolved ({url!r})"
+        if optional:
+            log.info("MCP server %r on agent %r dropped (optional): %s — continuing without it.",
+                     name, manifest.get("name"), reason)
         else:
-            dropped.add(s.get("name"))
+            log.warning("MCP server %r on agent %r dropped: %s — its tools will be MISSING until "
+                        "the URL is configured; re-provision after setting it.",
+                        name, manifest.get("name"), reason)
     tools = expand_env(manifest.get("tools", []))
     if dropped:
         tools = [t for t in tools

backend/core/package.json

@@ -10,6 +10,7 @@
     "test": "jest"
   },
   "dependencies": {
+    "@fuzefront/shared": "1.0.0",
     "cors": "^2.8.5",
     "express": "^4.19.2",
     "helmet": "^7.1.0",

backend/core/src/events/kafkaPublisher.ts

@@ -0,0 +1,172 @@
+import { Knex } from 'knex'
+import { ZodSchema } from 'zod'
+import {
+  createKafkaClient,
+  TypedProducer,
+  FuzeEvent,
+  dlqTopic,
+  schemaForTopic,
+  partitionKeyForPayload,
+} from '@fuzefront/shared/kafka'
+import {
+  OutboxRecord,
+  OutboxRelayHandle,
+  startOutboxRelay,
+} from './outboxRelay'
+
+export interface KafkaPublisherConfig {
+  brokers: string[]
+  clientId?: string
+}
+
+export interface KafkaOutboxPublisher {
+  /** Publish an outbox record to Kafka (validates via the shared schema registry). */
+  publish: (record: OutboxRecord) => Promise<void>
+  /** Route an exhausted record to its `<topic>.dlq`. */
+  deadLetter: (record: OutboxRecord) => Promise<void>
+  /** Disconnect the underlying producer (graceful shutdown). */
+  disconnect: () => Promise<void>
+}
+
+/** Minimal producer surface the publisher needs — satisfied by `TypedProducer`. */
+export interface ProducerLike {
+  send<T>(
+    topic: string,
+    event: FuzeEvent<T>,
+    schema: ZodSchema<T>,
+    options?: { key?: string }
+  ): Promise<void>
+  raw: { send(payload: { topic: string; messages: Array<{ key?: string; value: string }> }): Promise<unknown> }
+  disconnect(): Promise<void>
+}
+
+/**
+ * The generic transport wiring, decoupled from how the producer is obtained so
+ * it is unit-testable with a fake. Builds the `FuzeEvent` envelope, derives the
+ * partition key, and validates against the shared schema registry (unmapped
+ * topics publish raw). `getProducer` is called lazily/memoised by the caller.
+ */
+export function makeOutboxPublisher(getProducer: () => Promise<ProducerLike>): KafkaOutboxPublisher {
+  const publish = async (record: OutboxRecord): Promise<void> => {
+    // A connect/send failure throws → the relay leaves the row 'pending'.
+    const p = await getProducer()
+    const event: FuzeEvent<unknown> = {
+      version: '1.0',
+      topic: record.topic as FuzeEvent['topic'],
+      correlationId: record.correlationId,
+      occurredAt: new Date().toISOString(),
+      payload: record.payload,
+    }
+    const key = partitionKeyForPayload(record.payload)
+    const schema = schemaForTopic(record.topic)
+    if (schema) {
+      await p.send(record.topic, event, schema as ZodSchema<unknown>, { key })
+    } else {
+      await p.raw.send({
+        topic: record.topic,
+        messages: [{ key, value: JSON.stringify(event) }],
+      })
+    }
+  }
+
+  const deadLetter = async (record: OutboxRecord): Promise<void> => {
+    const p = await getProducer()
+    await p.raw.send({
+      topic: dlqTopic(record.topic),
+      messages: [
+        { value: JSON.stringify({ raw: record, reason: 'outbox max attempts exhausted' }) },
+      ],
+    })
+  }
+
+  const disconnect = async (): Promise<void> => {
+    // Only disconnect a producer that was actually created.
+    const p = await getProducer().catch(() => null)
+    if (p) await p.disconnect()
+  }
+
+  return { publish, deadLetter, disconnect }
+}
+
+/**
+ * Builds an outbox publisher backed by a lazily-connected Kafka `TypedProducer`.
+ */
+export function createKafkaOutboxPublisher(config: KafkaPublisherConfig): KafkaOutboxPublisher {
+  let producer: TypedProducer | null = null
+  let connecting: Promise<TypedProducer> | null = null
+
+  const getProducer = async (): Promise<ProducerLike> => {
+    if (producer) return producer
+    if (!connecting) {
+      connecting = (async () => {
+        const kafka = createKafkaClient({
+          clientId: config.clientId || 'fuzefront-outbox-relay',
+          brokers: config.brokers,
+        })
+        const p = new TypedProducer(kafka)
+        await p.connect()
+        producer = p
+        return p
+      })().catch(err => {
+        connecting = null // don't cache a failed connection
+        throw err
+      })
+    }
+    return connecting
+  }
+
+  const base = makeOutboxPublisher(getProducer)
+  return {
+    ...base,
+    disconnect: async () => {
+      if (producer) {
+        await producer.disconnect()
+        producer = null
+        connecting = null
+      }
+    },
+  }
+}
+
+export interface OutboxRelayFromEnvHandle extends OutboxRelayHandle {
+  disconnect: () => Promise<void>
+}
+
+/**
+ * Start the transactional-outbox relay with the Kafka transport wired from the
+ * environment — the one-call, install-and-go entry point for any backend
+ * service. Returns null (a no-op) when no broker is configured, so events stay
+ * durably in `event_outbox` until one is.
+ */
+export function startOutboxRelayFromEnv(opts: {
+  db: Knex
+  brokers?: string
+  clientId?: string
+  intervalMs?: number
+  logger?: { info: (m: string) => void; error: (m: string) => void }
+}): OutboxRelayFromEnvHandle | null {
+  const brokersRaw = opts.brokers ?? process.env.KAFKA_BROKERS
+  if (!brokersRaw) {
+    opts.logger?.info('KAFKA_BROKERS unset — outbox relay disabled (events held in event_outbox)')
+    return null
+  }
+  const brokers = brokersRaw
+    .split(',')
+    .map(b => b.trim())
+    .filter(Boolean)
+
+  const { publish, deadLetter, disconnect } = createKafkaOutboxPublisher({
+    brokers,
+    clientId: opts.clientId,
+  })
+
+  const handle = startOutboxRelay({
+    db: opts.db,
+    publish,
+    onDeadLetter: deadLetter,
+    intervalMs: opts.intervalMs ?? Number(process.env.OUTBOX_RELAY_INTERVAL_MS || 1000),
+    logger: opts.logger,
+  })
+
+  return { stop: handle.stop, disconnect }
+}

backend/core/src/index.ts

@@ -8,3 +8,4 @@ export * from './types/shared'
 export * from './bootstrap'
 export * from './events/outbox'
 export * from './events/outboxRelay'
+export * from './events/kafkaPublisher'