feat(slack): report local relay attempts in flight (#2569)

rig slack status adds "Relay attempts awaiting response: N" for the local Slack relay, beside the retained inbound records. The daemon's inbound router counts each forward to another host while it's awaiting a response (incremented before the send, decremented when it settles, success or not), and the connector status reports that number. It's point-in-time and covers only the current connector instance: after the Slack connector is restarted, a relay the previous instance started can still be awaiting its response (up to the forwarder's 10 s timeout) while the new count reads 0. Zero doesn't prove quiescence across hosts or instances. Routing, retries, retained records and the forwarder are unchanged.

docs/reference/slack-app-setup.md

@@ -349,8 +349,7 @@ decisions, and posts its own replies.
 - **Changing forwarding entries is not a drain or cancellation:** relays already retained keep
   their original host and remain visible in `rig slack status`, even after the last entry is
   removed. An enabled connector retries them; disabling it pauses retry without deleting them.
-  Let relays finish before removing a forward entry. A zero dead-letter backlog alone does not
-  prove they have drained: it does not count relay HTTP requests still in flight.
+  Let relays finish before removing a forward entry; `rig slack status` shows local in-flight attempts separately from retained records, but zero does not prove global quiescence.
 - **Who answered:** every post starts `From <seat> on <machine>`, using the machine's host id (set
   it with `host.name`).
 - **The hub's own channels** are handled on the hub, as before. A channel can be relayed to one

packages/cli/src/commands/slack.ts

@@ -252,7 +252,7 @@ export function slackCommand(deps: SlackDeps = {}): Command {
         const connector = observation.connector as { configurationDigest?: string; inboundReady?: boolean; outboundReady?: boolean;
           inboundRouting?: { fallback: string; destination: string };
           settlement?: { pending: number; lastFailure: string | null };
-          deadLetterBacklog?: number | null; deadLetterBacklogState?: string; deadLetterBacklogReason?: string;
+          deadLetterBacklog?: number | null; deadLetterBacklogState?: string; deadLetterBacklogReason?: string; relayAttemptsInFlight?: number;
           inbound?: InboundDeliveryStatus & { state?: string; generation?: number; lastEventAt?: string };
           recovery?: { state?: string; reason?: string; lastScanAt?: string; acceptedThisProcess?: number; deadLetteredThisProcess?: number;
             coverage?: { coverageStart: string; coveredThrough: string; pending?: { upper: string; nextLatest: string }; nextRetryAt?: number } | null;
@@ -291,6 +291,7 @@ export function slackCommand(deps: SlackDeps = {}): Command {
               + `${reasons(threads.backoff) ? `; backing off: ${reasons(threads.backoff)}` : ""}${reasons(threads.incomplete) ? `; incomplete: ${reasons(threads.incomplete)}` : ""}`);
           }
           log(`  Recovery counts since connector start: accepted ${recovery?.acceptedThisProcess ?? "unknown"}; dead-lettered ${recovery?.deadLetteredThisProcess ?? "unknown"} (custody, not delivery)`);
+          if (connector.relayAttemptsInFlight !== undefined) log(`  Relay attempts awaiting response: ${connector.relayAttemptsInFlight} (this process, point-in-time; zero does not prove global quiescence)`);
           if (connector.deadLetterBacklogState !== undefined) {
             const scope = "inbound messages, reactions and click answers";
             if (connector.deadLetterBacklogState === "unknown") {

packages/cli/test/slack-cutover-cli.test.ts

@@ -138,13 +138,15 @@ describe("S10 CLI cutover — config surfaces survive on the daemon-homed module
     { name: "enabled bot-only relay receiver", outboundReady: true, inboundReady: false, paused: false },
   ])("shows configured routing and retained retry state on a $name", async ({ outboundReady, inboundReady, paused }) => {
     const { deps, logs } = makeDeps({ clientFactory: () => ({ get: async <T>() => ({ status: 200,
-      data: { state: "active", connector: { outboundReady, inboundReady, inboundRouting: { fallback: "local", destination: "ops@hub" }, inbound: { state: "not-configured" }, deadLetterBacklog: 6, deadLetterBacklogState: "ok" } } as T }) }) });
+      data: { state: "active", connector: { outboundReady, inboundReady, inboundRouting: { fallback: "local", destination: "ops@hub" }, inbound: { state: "not-configured" }, deadLetterBacklog: 6, deadLetterBacklogState: "ok", relayAttemptsInFlight: 2 } } as T }) }) });
     await run(slackCommand(deps), ["status"]);
     expect(logs.join("\n")).toContain("Configured routing fallback: local to ops@hub (after forwarding, thread and channel routes)");
     expect(logs.join("\n")).toContain(paused ? "6 retained record(s); retry paused" : "6 retained record(s) awaiting retry");
+    expect(logs.join("\n")).toContain("Relay attempts awaiting response: 2 (this process, point-in-time; zero does not prove global quiescence)");
     logs.length = 0;
     await run(slackCommand(deps), ["status", "--json"]);
     expect(JSON.parse(logs[0]!).observation.connector.inboundRouting).toEqual({ fallback: "local", destination: "ops@hub" });
+    expect(JSON.parse(logs[0]!).observation.connector.relayAttemptsInFlight).toBe(2);
   });
 });
 

packages/daemon/src/domain/gateway/slack/inbound.ts

@@ -232,6 +232,7 @@ export interface InboundDeps {
 
 export class InboundRouter {
   private readonly inflight = new Set<string>(); // same-channel message identity double-dispatch guard
+  private relayAttemptsInFlight = 0;
   private readonly replyingTo = new Set<string>(); // click replies being posted now
   private retryPass: Promise<{ retried: number; landed: number }> | undefined;
   private readonly botApps = new Map<string, string>(); // bot_id → its app, from bots.info
@@ -247,6 +248,15 @@ export class InboundRouter {
     this.botLimiter = new BotLimiter(deps.now);
   }
 
+  /** Point-in-time local HTTP attempts, independent of durable retained records. */
+  getRelayAttemptsInFlight(): number { return this.relayAttemptsInFlight; }
+
+  private async sendForward(host: string, env: SocketEnvelope): Promise<Awaited<ReturnType<EnvelopeForwarder>>> {
+    this.relayAttemptsInFlight++;
+    try { return await this.deps.forward!.send(host, env); }
+    finally { this.relayAttemptsInFlight--; }
+  }
+
   /** The row's summary names the person who actually wrote (their registered address); its tags stay as they were. */
   private summaryOf(ev: SlackEvent, sender: string, transfer?: InboundFileResult | null, correlationQitemId?: string, envelope?: string): { summary: string; body: string; attachmentsChars?: number } {
     const text = String(ev.text ?? "");
@@ -909,7 +919,7 @@ export class InboundRouter {
   /** Relay an envelope to the machine that owns its channel. The hub has already acknowledged it to
    *  Slack, so a failed relay is kept with its host and retried, never dropped. */
   async forwardEnvelope(host: string, env: SocketEnvelope): Promise<{ status: InboundDisposition; reason?: string }> {
-    const r = await this.deps.forward!.send(host, env);
+    const r = await this.sendForward(host, env);
     const channel = envelopeChannel(env) ?? "-";
     if (r.ok) {
       this.deps.log?.(`forwarded ${env.type} channel=${channel} -> ${host}`);
@@ -936,7 +946,7 @@ export class InboundRouter {
     for (const e of entries) {
       if (failedHosts.has(e.ev.host)) { stillFailing.push(e); continue; }
       sent++;
-      const r = await this.deps.forward.send(e.ev.host, e.ev.envelope);
+      const r = await this.sendForward(e.ev.host, e.ev.envelope);
       if (r.ok) landed++;
       else { failedHosts.add(e.ev.host); stillFailing.push({ ev: e.ev, at: e.at, attempts: e.attempts + 1 }); }
     }

packages/daemon/src/domain/gateway/slack/slack-subsystem.ts

@@ -1213,6 +1213,7 @@ export function buildSlackGatewayWire(opts: SlackWireOpts): GatewayWire {
       // Durable inbound custody across event, click and relay files, even when inert.
       // Count records, not distinct messages; unreadable storage is null/unknown.
       ...deadLetterBacklogFields(dead, deadActions, deadForwards),
+      relayAttemptsInFlight: router?.getRelayAttemptsInFlight() ?? 0,
       // A machine with no connection of its own can take events a hub relays here; whether a hub does is
       // the hub's configuration, so this says only that it can, and how many have arrived.
       ...(router && !inboundReady ? { relayedInbound: { accepting: true, received: relayedReceived } } : {}),

packages/daemon/test/slack-shared-foundation-wire.test.ts

@@ -155,7 +155,7 @@ describe("one shared Slack app across two machines", () => {
   });
 
   it("reports the local fallback and still routes an unmapped channel to that destination", async () => {
-    expect(hub.wire!.status()).toMatchObject({ inboundRouting: { fallback: "local", destination: "orch-advisor@v-build" } });
+    expect(hub.wire!.status()).toMatchObject({ inboundRouting: { fallback: "local", destination: "orch-advisor@v-build" }, relayAttemptsInFlight: 0 });
     await say("C0UNMAPPED", "735.1");
     await vi.waitFor(() => expect(inbound(hub)).toHaveLength(1));
     expect(inbound(hub)[0]!.destinationSession).toBe("orch-advisor@v-build");

packages/daemon/test/slack-shared-foundation.test.ts

@@ -2,7 +2,7 @@
 // rig's representative or a configured seat; a person's top-level message becomes the root its answer
 // lands under; the hub relays a channel another machine owns, keeps a failed relay and retries it;
 // the receiving route takes a relayed envelope; and every post names the machine it comes from.
-import { describe, it, expect } from "vitest";
+import { describe, it, expect, vi } from "vitest";
 import { Hono } from "hono";
 import { createDb } from "../src/db/connection.js";
 import { migrate } from "../src/db/migrate.js";
@@ -165,6 +165,31 @@ describe("the hub relays a channel another machine owns", () => {
   }
   const message = (channel: string): SocketEnvelope => ({ envelope_id: "e1", type: "events_api", payload: { event: { type: "message", user: "U1", text: "hi", ts: "700.1", channel } } });
 
+  it("counts only pending live and retry relay attempts, including failed sends", async () => {
+    let finish!: (result: { ok: false; error: string } | { ok: true }) => void;
+    const h = hub(() => new Promise((resolve) => { finish = resolve; }));
+    const live = handleEnvelope(message("C0VPS04"), () => {}, h.r);
+    await vi.waitFor(() => expect(h.r.getRelayAttemptsInFlight()).toBe(1));
+    expect(h.forwardDeadLetter.readAll()).toHaveLength(0);
+    finish({ ok: false, error: "synthetic failure" });
+    await live;
+    expect(h.r.getRelayAttemptsInFlight()).toBe(0);
+    expect(h.forwardDeadLetter.readAll()).toHaveLength(1);
+    const retry = h.r.retryDeadLetters();
+    await vi.waitFor(() => expect(h.r.getRelayAttemptsInFlight()).toBe(1));
+    expect(h.forwardDeadLetter.readAll()).toHaveLength(1);
+    finish({ ok: true });
+    await retry;
+    expect(h.r.getRelayAttemptsInFlight()).toBe(0);
+    expect(h.forwardDeadLetter.readAll()).toHaveLength(0);
+  });
+
+  it("clears the attempt count when the transport throws", async () => {
+    const h = hub(async () => { throw new Error("synthetic error"); });
+    await expect(h.r.forwardEnvelope("vps04", message("C0VPS04"))).rejects.toThrow("synthetic error");
+    expect(h.r.getRelayAttemptsInFlight()).toBe(0);
+  });
+
   it("sends a message, a reaction and a click for that channel there, untouched, and handles nothing here", async () => {
     const sent: Array<[string, SocketEnvelope]> = [];
     const h = hub(async (host, env) => { sent.push([host, env]); return { ok: true }; });