diff --git a/src/components/journal/index.ts b/src/components/journal/index.ts
new file mode 100644
index 0000000..c1737f3
--- /dev/null
+++ b/src/components/journal/index.ts
@@ -0,0 +1,35 @@
+/**
+ * Journal module — barrel export
+ */
+
+// Trigger message type registration
+import "./types";
+
+export {
+  registerMessageType,
+  getMessageType,
+  registeredTypes,
+  validatePayload,
+  canEmit,
+} from "./registry";
+export type { MessageTypeDef, StreamMessage } from "./registry";
+export {
+  connectStream,
+  disconnectStream,
+  hydrateFromServer,
+  sendMessage,
+  revertLatest,
+  canRevert,
+  visibleMessages,
+  useJournalStream,
+  journalStreamState,
+  createSession,
+  deleteSession,
+  sessions,
+  setMyName,
+} from "../stores/journalStream";
+export type {
+  JournalStreamState,
+  SessionMeta,
+  SessionManifest,
+} from "../stores/journalStream";
diff --git a/src/components/journal/registry.ts b/src/components/journal/registry.ts
new file mode 100644
index 0000000..ba0d9d5
--- /dev/null
+++ b/src/components/journal/registry.ts
@@ -0,0 +1,126 @@
+/**
+ * Journal Stream — Message Type Registry
+ *
+ * An open, add-only registry of message types. Game-specific widgets
+ * call `registerMessageType()` at module import time. The stream core
+ * uses this to validate, render, and reduce messages.
+ */
+
+import type { ZodType } from "zod";
+import type { JSX } from "solid-js";
+
+// ---------------------------------------------------------------------------
+// StreamMessage
+// ---------------------------------------------------------------------------
+
+/**
+ * The typed envelope carried on the stream topic.
+ * `P` is narrowed by the `type` key at the receiving side.
+ */
+export interface StreamMessage<P = unknown> {
+  /** Deterministic id: `{sender}-{seq}` */
+  id: string;
+  /** Sender identifier — "gm" or a player name */
+  sender: string;
+  /** Monotonic sequence number, per sender */
+  seq: number;
+  /** Registered type key — determines which payload schema & widget are used */
+  type: string;
+  /** Type-specific data, validated against the registry's Zod schema */
+  payload: P;
+  /** Epoch millis, set by the sender at publish time */
+  timestamp: number;
+  /**
+   * Soft-delete. Only the sender's latest (highest seq) message can be
+   * reverted. Sending a new message locks the previous one permanently.
+   */
+  reverted: boolean;
+}
+
+// ---------------------------------------------------------------------------
+// MessageTypeDef
+// ---------------------------------------------------------------------------
+
+export interface MessageTypeDef<P = unknown> {
+  /** Unique type key, e.g. "roll.request", "article.reveal" */
+  type: string;
+  /** Human-readable label shown in the compose panel dropdown */
+  label: string;
+  /** Which roles can emit this type. Empty array = all roles allowed. */
+  emitters?: ("gm" | "player")[];
+  /** Zod schema used to validate payload before publish */
+  schema: ZodType<P>;
+  /** Returns a default payload for compose panel initialization */
+  defaultPayload?: () => P;
+  /**
+   * Render the message body in the stream.
+   * Receives the full message for context (sender, timestamp, reverted).
+   * Return undefined to suppress inline rendering (e.g. pure state mutations).
+   */
+  render?: (payload: P, msg: StreamMessage<P>) => JSX.Element | undefined;
+  /**
+   * Apply side-effects to local state when this message is received.
+   * Runs once per message, at initial hydration and on live receipt.
+   */
+  reducer?: (payload: P, msg: StreamMessage<P>) => void;
+}
+
+// ---------------------------------------------------------------------------
+// Registry
+// ---------------------------------------------------------------------------
+
+const registry = new Map<string, MessageTypeDef<any>>();
+
+/**
+ * Register a message type. Calling module should import and call this at
+ * top-level so the registry is populated before any stream operations.
+ */
+export function registerMessageType<P>(def: MessageTypeDef<P>): void {
+  if (registry.has(def.type)) {
+    throw new Error(`Message type "${def.type}" is already registered.`);
+  }
+  registry.set(def.type, def);
+}
+
+/** Retrieve a registered type definition. */
+export function getMessageType(type: string): MessageTypeDef | undefined {
+  return registry.get(type);
+}
+
+/** Iterate over all registered type keys. */
+export function registeredTypes(): IterableIterator<string> {
+  return registry.keys();
+}
+
+/**
+ * Validate a payload against the registered schema for `type`.
+ * Returns the parsed (narrowed) payload on success, or a ZodError.
+ */
+export function validatePayload<P>(
+  type: string,
+  payload: P,
+): { success: true; data: P } | { success: false; error: string } {
+  const def = registry.get(type);
+  if (!def) {
+    return { success: false, error: `Unknown message type: "${type}"` };
+  }
+  const result = def.schema.safeParse(payload);
+  if (result.success) {
+    return { success: true, data: result.data as P };
+  }
+  return {
+    success: false,
+    error: result.error.issues.map((i) => `${i.path.join(".")}: ${i.message}`).join("; "),
+  };
+}
+
+/**
+ * Check whether a role (sender) is allowed to emit a given message type.
+ */
+export function canEmit(type: string, role: string): boolean {
+  const def = registry.get(type);
+  if (!def) return false;
+  const emitters = def.emitters;
+  if (!emitters || emitters.length === 0) return true;
+  return emitters.includes(role as "gm" | "player");
+}
diff --git a/src/components/journal/types/article.ts b/src/components/journal/types/article.ts
new file mode 100644
index 0000000..b6a1d62
--- /dev/null
+++ b/src/components/journal/types/article.ts
@@ -0,0 +1,37 @@
+/**
+ * Built-in message type: article.reveal
+ *
+ * GM reveals a document/article path to players. The reducer populates
+ * the revealedPaths set so the Article component can gate visibility.
+ */
+
+import { z } from "zod";
+import { registerMessageType } from "../registry";
+import { journalSetState } from "../../stores/journalStream";
+import { produce } from "solid-js/store";
+
+const schema = z.object({
+  /** Path relative to content root, e.g. "/adventures/dungeon.md" */
+  path: z.string().min(1),
+  /** Optional display label, defaults to the path */
+  label: z.string().optional(),
+  /** Optional section heading to deep-link */
+  section: z.string().optional(),
+});
+
+export type ArticleRevealPayload = z.infer<typeof schema>;
+
+registerMessageType<ArticleRevealPayload>({
+  type: "article.reveal",
+  label: "Reveal Article",
+  emitters: ["gm"],
+  schema,
+  defaultPayload: () => ({ path: "/" }),
+  reducer: (p) => {
+    journalSetState(
+      produce((s) => {
+        s.revealedPaths.add(p.path);
+      }),
+    );
+  },
+});
diff --git a/src/components/journal/types/index.ts b/src/components/journal/types/index.ts
new file mode 100644
index 0000000..3ff6a0e
--- /dev/null
+++ b/src/components/journal/types/index.ts
@@ -0,0 +1,11 @@
+/**
+ * Journal message types — barrel export
+ *
+ * Importing this module registers all built-in message types.
+ * Game-specific widgets should create their own barrel and import it.
+ */
+
+import "./narrative";
+import "./roll";
+import "./article";
+import "./intent";
diff --git a/src/components/journal/types/intent.ts b/src/components/journal/types/intent.ts
new file mode 100644
index 0000000..15772d2
--- /dev/null
+++ b/src/components/journal/types/intent.ts
@@ -0,0 +1,58 @@
+/**
+ * Built-in message types: intent / resolution
+ *
+ * intent      — Player declares an action or resource usage for GM to resolve
+ * resolution  — GM resolves a player intent (outcome, stat changes, etc.)
+ */
+
+import { z } from "zod";
+import { registerMessageType } from "../registry";
+
+// ---------------------------------------------------------------------------
+// intent
+// ---------------------------------------------------------------------------
+
+const intentSchema = z.object({
+  /** Plain-text description of what the player wants to do */
+  description: z.string().min(1),
+  /**
+   * Optional JSON-serializable context (item name, skill name, target, etc.)
+   * Parsed by game-specific widgets.
+   */
+  context: z.record(z.string(), z.unknown()).optional(),
+});
+
+export type IntentPayload = z.infer<typeof intentSchema>;
+
+registerMessageType<IntentPayload>({
+  type: "intent",
+  label: "Declare Intent",
+  emitters: ["player"],
+  schema: intentSchema,
+  defaultPayload: () => ({ description: "" }),
+});
+
+// ---------------------------------------------------------------------------
+// resolution
+// ---------------------------------------------------------------------------
+
+const resolutionSchema = z.object({
+  /** The id of the intent message being resolved */
+  intentId: z.string().min(1),
+  /** Freeform outcome text */
+  outcome: z.string().min(1),
+  /**
+   * Stat changes to apply. Game-specific widgets interpret these keys.
+   * Example: { "hp": -3, "stress": 1, "status.add": "bleeding" }
+   */
+  statChanges: z.record(z.string(), z.unknown()).optional(),
+});
+
+export type ResolutionPayload = z.infer<typeof resolutionSchema>;
+
+registerMessageType<ResolutionPayload>({
+  type: "resolution",
+  label: "Resolve Intent",
+  emitters: ["gm"],
+  schema: resolutionSchema,
+});
diff --git a/src/components/journal/types/narrative.ts b/src/components/journal/types/narrative.ts
new file mode 100644
index 0000000..26b48b5
--- /dev/null
+++ b/src/components/journal/types/narrative.ts
@@ -0,0 +1,26 @@
+/**
+ * Built-in message type: narrative
+ *
+ * Freeform text from GM or players. Renders as a chat bubble.
+ */
+
+import { z } from "zod";
+import { registerMessageType } from "../registry";
+
+const schema = z.object({
+  text: z.string().min(1, "Message cannot be empty").max(2000, "Message too long"),
+});
+
+export type NarrativePayload = z.infer<typeof schema>;
+
+registerMessageType<NarrativePayload>({
+  type: "narrative",
+  label: "Narrative",
+  schema,
+  defaultPayload: () => ({ text: "" }),
+  render: (p, msg) => {
+    // Placeholder — real render will be in StreamMessage component
+    // Returning a simple representation for now
+    return undefined;
+  },
+});
diff --git a/src/components/journal/types/roll.ts b/src/components/journal/types/roll.ts
new file mode 100644
index 0000000..f02ee3a
--- /dev/null
+++ b/src/components/journal/types/roll.ts
@@ -0,0 +1,60 @@
+/**
+ * Built-in message types: roll.request / roll.result
+ *
+ * roll.request  — GM asks for a roll, renders a prompt with dice roller
+ * roll.result   — Player responds with their roll outcome
+ */
+
+import { z } from "zod";
+import { registerMessageType } from "../registry";
+
+// ---------------------------------------------------------------------------
+// roll.request
+// ---------------------------------------------------------------------------
+
+const rollRequestSchema = z.object({
+  /** Dice notation, e.g. "2d6", "1d20+5", "3d6kh1" */
+  notation: z.string().min(1),
+  /** Human label, e.g. "Fear save", "Attack roll" */
+  label: z.string().min(1),
+  /** Optional target number to compare against */
+  target: z.number().int().optional(),
+  /** Optional extra context */
+  description: z.string().optional(),
+});
+
+export type RollRequestPayload = z.infer<typeof rollRequestSchema>;
+
+registerMessageType<RollRequestPayload>({
+  type: "roll.request",
+  label: "Request Roll",
+  emitters: ["gm"],
+  schema: rollRequestSchema,
+  defaultPayload: () => ({ notation: "1d20", label: "" }),
+});
+
+// ---------------------------------------------------------------------------
+// roll.result
+// ---------------------------------------------------------------------------
+
+const rollResultSchema = z.object({
+  /** The dice notation that was rolled */
+  notation: z.string().min(1),
+  /** The raw dice results, e.g. [6, 4] for 2d6 */
+  dice: z.array(z.number().int().min(1)),
+  /** Modifiers, e.g. +5 */
+  modifier: z.number().int().default(0),
+  /** Final total after modifiers and keep/drop rules */
+  total: z.number(),
+  /** Optional: which roll.request message this is responding to (by id) */
+  requestId: z.string().optional(),
+});
+
+export type RollResultPayload = z.infer<typeof rollResultSchema>;
+
+registerMessageType<RollResultPayload>({
+  type: "roll.result",
+  label: "Roll Result",
+  emitters: ["player"],
+  schema: rollResultSchema,
+});
diff --git a/src/components/stores/journalStream.ts b/src/components/stores/journalStream.ts
new file mode 100644
index 0000000..bb44ad4
--- /dev/null
+++ b/src/components/stores/journalStream.ts
@@ -0,0 +1,441 @@
+/**
+ * Journal Stream — Client Store
+ *
+ * Reactive state for a single session's message stream. Manages MQTT
+ * connection, local message log, per-sender sequence tracking, and the
+ * revealed-paths set (populated by the article.reveal reducer).
+ *
+ * Session lifecycle (create/list/delete) is handled via MQTT retained
+ * topics — ttrpg/$SESSIONS for the manifest and ttrpg/{id}/meta per session.
+ *
+ * No persistence here — that's the CLI server's job via JSONL append.
+ */
+
+import { createStore, produce } from "solid-js/store";
+import { createSignal } from "solid-js";
+import type { StreamMessage } from "../journal/registry";
+import { getMessageType, validatePayload } from "../journal/registry";
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+export interface JournalStreamState {
+  sessionId: string | null;
+  /** Full message log, oldest-first */
+  messages: StreamMessage[];
+  /** Last sequence number per sender */
+  senderSeq: Record<string, number>;
+  /**
+   * Paths revealed by article.reveal messages.
+   * Populated during hydration and live receipt via the type's reducer.
+   */
+  revealedPaths: Set<string>;
+  /** MQTT connection status */
+  connected: boolean;
+  /** This client's identity */
+  myName: string;
+  /** Broker URL, set after connect */
+  brokerUrl: string | null;
+}
+
+export interface SessionMeta {
+  name: string;
+  created: number;
+  players: string[];
+}
+
+export interface SessionManifest {
+  sessions: Record<string, SessionMeta>;
+}
+
+// ---------------------------------------------------------------------------
+// localStorage keys
+// ---------------------------------------------------------------------------
+
+const LS_PLAYER_NAME = "ttrpg.playerName";
+const LS_BROKER_URL = "ttrpg.brokerUrl";
+const LS_LAST_SESSION = "ttrpg.lastSessionId";
+
+function loadPersisted(): {
+  myName: string;
+  brokerUrl: string | null;
+  lastSessionId: string | null;
+} {
+  if (typeof localStorage === "undefined") {
+    return { myName: "gm", brokerUrl: null, lastSessionId: null };
+  }
+  return {
+    myName: localStorage.getItem(LS_PLAYER_NAME) || "gm",
+    brokerUrl: localStorage.getItem(LS_BROKER_URL),
+    lastSessionId: localStorage.getItem(LS_LAST_SESSION),
+  };
+}
+
+// ---------------------------------------------------------------------------
+// Store
+// ---------------------------------------------------------------------------
+
+const persisted = loadPersisted();
+
+const [state, setState] = createStore<JournalStreamState>({
+  sessionId: persisted.lastSessionId,
+  messages: [],
+  senderSeq: {},
+  revealedPaths: new Set(),
+  connected: false,
+  myName: persisted.myName,
+  brokerUrl: persisted.brokerUrl,
+});
+
+export { setState as journalSetState };
+
+const [sessionList, setSessionList] = createSignal<SessionManifest>({
+  sessions: {},
+});
+
+export { sessionList as sessions };
+
+/**
+ * Change the current player's name. Persisted to localStorage so it
+ * survives page reloads.
+ */
+export function setMyName(name: string): void {
+  setState("myName", name);
+  if (typeof localStorage !== "undefined") {
+    localStorage.setItem(LS_PLAYER_NAME, name);
+  }
+}
+
+// Will hold the MQTT client instance after connect()
+let _mqttClient: import("mqtt").MqttClient | null = null;
+let _mqttConnected = false;
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function makeMessageId(sender: string, seq: number): string {
+  return `${sender}-${seq}`;
+}
+
+function runReducer(msg: StreamMessage): void {
+  const def = getMessageType(msg.type);
+  if (def?.reducer) {
+    def.reducer(msg.payload, msg);
+  }
+}
+
+const $SESSIONS = "ttrpg/$SESSIONS";
+
+// ---------------------------------------------------------------------------
+// Hydration (initial load from server)
+// ---------------------------------------------------------------------------
+
+/**
+ * Load the full message history from the static server's JSONL file.
+ * The file is served from the same HTTP origin as the web app.
+ * Runs all reducers in order.
+ */
+export async function hydrateFromServer(sessionId: string): Promise<void> {
+  const response = await fetch(
+    `/.ttrpg/sessions/${encodeURIComponent(sessionId)}/stream.jsonl`,
+  );
+
+  if (!response.ok) {
+    if (response.status === 404) {
+      return; // fresh session, no file yet
+    }
+    throw new Error(`Failed to load session: ${response.statusText}`);
+  }
+
+  const text = await response.text();
+  const lines = text.split("\n").filter((l) => l.trim());
+
+  const messages: StreamMessage[] = [];
+  const senderSeq: Record<string, number> = {};
+
+  for (const line of lines) {
+    try {
+      const msg: StreamMessage = JSON.parse(line);
+      messages.push(msg);
+      senderSeq[msg.sender] = Math.max(senderSeq[msg.sender] ?? 0, msg.seq);
+      runReducer(msg);
+    } catch {
+      /* skip corrupt */
+    }
+  }
+
+  setState(
+    produce((s) => {
+      s.messages = messages;
+      s.senderSeq = senderSeq;
+      s.sessionId = sessionId;
+    }),
+  );
+}
+
+// ---------------------------------------------------------------------------
+// MQTT Connect
+// ---------------------------------------------------------------------------
+
+/**
+ * Connect to the MQTT broker, subscribe to the session stream, session
+ * list, and session meta. Must be called after `hydrateFromServer`.
+ */
+export async function connectStream(
+  sessionId: string,
+  brokerUrl: string,
+): Promise<void> {
+  const { default: mqtt } = await import("mqtt");
+
+  const client = mqtt.connect(brokerUrl, {
+    clientId: `${state.myName}-${Date.now()}`,
+    protocol: brokerUrl.startsWith("wss") ? "wss" : "ws",
+    reconnectPeriod: 2000,
+  });
+
+  _mqttClient = client;
+
+  client.on("connect", () => {
+    _mqttConnected = true;
+    setState("connected", true);
+    setState("brokerUrl", brokerUrl);
+
+    // Persist connection info for next time
+    if (typeof localStorage !== "undefined") {
+      localStorage.setItem(LS_BROKER_URL, brokerUrl);
+      localStorage.setItem(LS_LAST_SESSION, sessionId);
+    }
+
+    client.subscribe(`ttrpg/${sessionId}/stream`, { qos: 1 }, (err) => {
+      if (err) console.error("[stream] stream sub err:", err);
+    });
+    client.subscribe($SESSIONS, { qos: 1 }, (err) => {
+      if (err) console.error("[stream] sessions sub err:", err);
+    });
+    client.subscribe(`ttrpg/${sessionId}/meta`, { qos: 1 });
+  });
+
+  client.on("message", (topic, payload) => {
+    const raw = payload.toString();
+    const parts = topic.split("/");
+
+    if (topic === $SESSIONS) {
+      // Session manifest update
+      try {
+        const manifest: SessionManifest = JSON.parse(raw);
+        setSessionList(manifest);
+      } catch (e) {
+        console.error("[stream] manifest parse err:", e);
+      }
+      return;
+    }
+
+    if (parts.length >= 3 && parts[0] === "ttrpg" && parts[2] === "meta") {
+      // Session meta update — manifest will be republished by server,
+      // handled via $SESSIONS subscription above.
+      return;
+    }
+
+    if (parts.length >= 3 && parts[0] === "ttrpg" && parts[2] === "stream") {
+      try {
+        const msg: StreamMessage = JSON.parse(raw);
+        receiveMessage(msg);
+      } catch (e) {
+        console.error("[stream] malformed message:", e);
+      }
+    }
+  });
+
+  client.on("close", () => {
+    _mqttConnected = false;
+    setState("connected", false);
+  });
+
+  client.on("error", (err) => {
+    console.error("[stream] mqtt error:", err);
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Session lifecycle
+// ---------------------------------------------------------------------------
+
+/**
+ * Create a new session by publishing retained metadata to its meta topic.
+ * The server picks it up and adds it to the $SESSIONS manifest.
+ */
+export function createSession(name: string, players: string[] = []): void {
+  if (!_mqttClient || !_mqttConnected) return;
+
+  const id =
+    name
+      .toLowerCase()
+      .replace(/[^a-z0-9]+/g, "-")
+      .replace(/^-|-$/g, "") || "session";
+
+  const meta: SessionMeta = { name, created: Date.now(), players };
+
+  _mqttClient.publish(`ttrpg/${id}/meta`, JSON.stringify(meta), {
+    qos: 1,
+    retain: true,
+  });
+}
+
+/**
+ * Delete a session: publish tombstone (empty payload) to its meta topic.
+ */
+export function deleteSession(sessionId: string): void {
+  if (!_mqttClient || !_mqttConnected) return;
+
+  _mqttClient.publish(`ttrpg/${sessionId}/meta`, "", { qos: 1, retain: true });
+}
+
+// ---------------------------------------------------------------------------
+// Send
+// ---------------------------------------------------------------------------
+
+/**
+ * Publish a message to the stream. Validates the payload against the
+ * registered Zod schema before sending. Auto-increments the sender's seq.
+ */
+export function sendMessage<T>(
+  type: string,
+  payload: T,
+):
+  { success: true; msg: StreamMessage<T> } | { success: false; error: string } {
+  if (!_mqttClient || !_mqttConnected) {
+    return { success: false, error: "Not connected to stream" };
+  }
+
+  const sessionId = state.sessionId;
+  if (!sessionId) {
+    return { success: false, error: "No active session" };
+  }
+
+  const validation = validatePayload(type, payload);
+  if (!validation.success) {
+    return { success: false, error: validation.error };
+  }
+
+  const sender = state.myName;
+  const seq = (state.senderSeq[sender] ?? 0) + 1;
+  const id = makeMessageId(sender, seq);
+
+  const msg: StreamMessage<T> = {
+    id,
+    sender,
+    seq,
+    type,
+    payload: validation.data,
+    timestamp: Date.now(),
+    reverted: false,
+  };
+
+  const topic = `ttrpg/${sessionId}/stream`;
+  _mqttClient.publish(topic, JSON.stringify(msg), { qos: 1 }, (err) => {
+    if (err) console.error("[stream] publish error:", err);
+  });
+
+  // Optimistic local insert
+  receiveMessage(msg as StreamMessage);
+
+  return { success: true, msg };
+}
+
+// ---------------------------------------------------------------------------
+// Receive
+// ---------------------------------------------------------------------------
+
+function receiveMessage(msg: StreamMessage): void {
+  const existing = state.messages.find((m) => m.id === msg.id);
+  if (existing) {
+    setState(
+      produce((s) => {
+        const idx = s.messages.findIndex((m) => m.id === msg.id);
+        if (idx !== -1) s.messages[idx] = msg;
+      }),
+    );
+    return;
+  }
+
+  setState(
+    produce((s) => {
+      s.messages.push(msg);
+      s.senderSeq[msg.sender] = Math.max(s.senderSeq[msg.sender] ?? 0, msg.seq);
+    }),
+  );
+
+  runReducer(msg);
+}
+
+// ---------------------------------------------------------------------------
+// Revert
+// ---------------------------------------------------------------------------
+
+/**
+ * Revert the current sender's latest (highest seq) message.
+ * Only works if it's still the latest — a subsequent message locks it.
+ */
+export function revertLatest():
+  { success: true } | { success: false; error: string } {
+  if (!_mqttClient || !_mqttConnected) {
+    return { success: false, error: "Not connected to stream" };
+  }
+
+  const sessionId = state.sessionId;
+  if (!sessionId) return { success: false, error: "No active session" };
+
+  const sender = state.myName;
+  const latestSeq = state.senderSeq[sender];
+  if (!latestSeq) return { success: false, error: "No messages to revert" };
+
+  const id = makeMessageId(sender, latestSeq);
+  const original = state.messages.find((m) => m.id === id);
+  if (!original) return { success: false, error: "Message not found" };
+  if (original.reverted) return { success: false, error: "Already reverted" };
+
+  const reverted: StreamMessage = { ...original, reverted: true };
+  const topic = `ttrpg/${sessionId}/stream`;
+  _mqttClient.publish(topic, JSON.stringify(reverted), { qos: 1 }, (err) => {
+    if (err) console.error("[stream] revert publish error:", err);
+  });
+
+  return { success: true };
+}
+
+// ---------------------------------------------------------------------------
+// Disconnect
+// ---------------------------------------------------------------------------
+
+export function disconnectStream(): void {
+  if (_mqttClient) {
+    _mqttClient.end(true);
+    _mqttClient = null;
+    _mqttConnected = false;
+  }
+  setState("connected", false);
+}
+
+// ---------------------------------------------------------------------------
+// Derived / helpers
+// ---------------------------------------------------------------------------
+
+export function canRevert(): boolean {
+  const sender = state.myName;
+  const seq = state.senderSeq[sender];
+  if (!seq) return false;
+  const msg = state.messages.find((m) => m.id === makeMessageId(sender, seq));
+  return msg !== undefined && !msg.reverted;
+}
+
+export function visibleMessages(): StreamMessage[] {
+  return state.messages.filter((m) => !m.reverted);
+}
+
+export function useJournalStream() {
+  return state;
+}
+
+export { state as journalStreamState };