fix(agent-presence): upsert atomically — kill first-time-insert race

Previous setStatus() did read-modify-write:
  findOne → if-exists save / else create+save

Two concurrent first-time writes for the same userId both saw no row,
both INSERT'd, second hit unique-key (agent_presences.PRIMARY) and 500'd
with "Duplicate entry '<userId>' for key 'agent_presences.PRIMARY'" —
visible in prod (2026-05-25 23:23:35Z) when Fabric.OpenclawPlugin's
presence-sync emitted two PUTs ~10 ms apart for the same agent (its
tick-overlap is being fixed separately in nav/Fabric.OpenclawPlugin).

Replace with repo.upsert(values, ['userId']) — compiles to MySQL
`INSERT … ON DUPLICATE KEY UPDATE`, atomic at the storage engine,
no read needed, no race window. Synthesize the returned entity from
the values we just wrote rather than a SELECT round-trip; controller
only reads {userId, status} off it.

Sim verified with 5 parallel PUTs to a fresh userId: all 200, no
Duplicate errors in guild log (was: 1 × 200 + 4 × 500 with the
old code).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

package.json

@@ -7,6 +7,7 @@
   "scripts": {
     "build": "tsc -p tsconfig.build.json",
     "start": "node dist/main.js",
+    "print:commands-key": "node dist/cli/print-commands-sync-key.js",
     "start:dev": "ts-node src/main.ts",
     "lint": "eslint 'src/**/*.ts'",
     "lint:fix": "eslint 'src/**/*.ts' --fix",

src/agents/agent-presence.controller.ts

@@ -0,0 +1,42 @@
+import { BadRequestException, Body, Controller, Get, Param, Put } from '@nestjs/common';
+import { AgentPresenceService, PresenceStatus } from './agent-presence.service.js';
+
+const VALID: PresenceStatus[] = ['idle', 'on_call', 'busy', 'exhausted', 'offline', 'unknown'];
+
+interface PutBody {
+  status?: string;
+  source?: string;
+}
+
+@Controller('agents/:userId/presence')
+export class AgentPresenceController {
+  constructor(private readonly svc: AgentPresenceService) {}
+
+  /**
+   * Read a user's current presence cache row.
+   * Auth: ApiKeyGuard (global). Any introspected center user can read.
+   */
+  @Get()
+  async get(@Param('userId') userId: string): Promise<{ userId: string; status: PresenceStatus }> {
+    const status = await this.svc.getStatus(userId);
+    return { userId, status };
+  }
+
+  /**
+   * Push a presence update. Called by Fabric.OpenclawPlugin's
+   * `presence-sync` loop on each delta. Auth: ApiKeyGuard (global) +
+   * the plugin uses its center-introspected api key.
+   *
+   * `source` is a debug tag describing who pushed (e.g. 'hf-plugin',
+   * 'manual'). Stored verbatim for trail.
+   */
+  @Put()
+  async put(@Param('userId') userId: string, @Body() body: PutBody): Promise<{ userId: string; status: PresenceStatus }> {
+    if (!body?.status || !VALID.includes(body.status as PresenceStatus)) {
+      throw new BadRequestException(`status must be one of ${VALID.join('|')}`);
+    }
+    const source = (body.source ?? 'unknown').slice(0, 64);
+    const row = await this.svc.setStatus(userId, body.status as PresenceStatus, source);
+    return { userId: row.userId, status: row.status };
+  }
+}

src/agents/agent-presence.module.ts

@@ -0,0 +1,13 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { AgentPresence } from '../entities/agent-presence.entity.js';
+import { AgentPresenceController } from './agent-presence.controller.js';
+import { AgentPresenceService } from './agent-presence.service.js';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([AgentPresence])],
+  controllers: [AgentPresenceController],
+  providers: [AgentPresenceService],
+  exports: [AgentPresenceService],
+})
+export class AgentPresenceModule {}

src/agents/agent-presence.service.ts

@@ -0,0 +1,61 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import { AgentPresence } from '../entities/agent-presence.entity.js';
+
+export type PresenceStatus = 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
+
+@Injectable()
+export class AgentPresenceService {
+  constructor(
+    @InjectRepository(AgentPresence)
+    private readonly repo: Repository<AgentPresence>,
+  ) {}
+
+  /**
+   * Get a user's current presence. Returns 'unknown' if no row.
+   * Used by `RealtimeGateway` per-recipient when xType === 'announce'.
+   */
+  async getStatus(userId: string): Promise<PresenceStatus> {
+    if (!userId) return 'unknown';
+    const row = await this.repo.findOne({ where: { userId } });
+    return row?.status ?? 'unknown';
+  }
+
+  /** Bulk variant for delivery-time lookups across many recipients in one trip. */
+  async getStatusMap(userIds: string[]): Promise<Map<string, PresenceStatus>> {
+    const out = new Map<string, PresenceStatus>();
+    for (const id of userIds) out.set(id, 'unknown');
+    if (userIds.length === 0) return out;
+    const rows = await this.repo
+      .createQueryBuilder('p')
+      .where('p.userId IN (:...ids)', { ids: userIds })
+      .getMany();
+    for (const r of rows) out.set(r.userId, r.status);
+    return out;
+  }
+
+  /**
+   * Upsert a user's presence. Source is a free-text tag for debugging
+   * (e.g. "hf-plugin", "manual", "test"). PUT /agents/:id/presence
+   * calls this; the plugin pushes only on diff so writes are sparse.
+   *
+   * Implementation note: the older findOne+save split was a read-modify-
+   * write race — two concurrent first-time writes for the same userId
+   * would both read no row, both INSERT, second hits unique-key dup
+   * (`agent_presences.PRIMARY`) and 500s. Fabric.OpenclawPlugin's
+   * presence-sync occasionally fires two PUTs for the same agent within
+   * ~10 ms (tick overlap on its side — separate fix in the plugin),
+   * which surfaced this race in prod.
+   *
+   * `repo.upsert(values, conflictPaths)` compiles to MySQL
+   * `INSERT … ON DUPLICATE KEY UPDATE` and is atomic at the storage
+   * engine level — no read needed, no race window. We synthesize the
+   * returned entity from what we just wrote rather than round-tripping
+   * a SELECT — the controller only reads {userId, status} off it.
+   */
+  async setStatus(userId: string, status: PresenceStatus, source: string): Promise<AgentPresence> {
+    await this.repo.upsert({ userId, status, source }, ['userId']);
+    return this.repo.create({ userId, status, source });
+  }
+}

src/app.module.ts

@@ -16,6 +16,7 @@ import { MembersModule } from './members/members.module.js';
 import { FilesModule } from './files/files.module.js';
 import { CanvasModule } from './canvas/canvas.module.js';
 import { CommandsModule } from './commands/commands.module.js';
+import { AgentPresenceModule } from './agents/agent-presence.module.js';
 
 @Module({
   imports: [
@@ -30,6 +31,7 @@ import { CommandsModule } from './commands/commands.module.js';
     FilesModule,
     CanvasModule,
     CommandsModule,
+    AgentPresenceModule,
   ],
   controllers: [HealthController, MetricsController],
   providers: [

src/channels/channels.controller.ts

@@ -1,4 +1,4 @@
-import { Body, Controller, Get, Param, Post, Query, Req, UnauthorizedException } from '@nestjs/common';
+import { BadRequestException, Body, Controller, Get, Param, Patch, Post, Query, Req, UnauthorizedException } from '@nestjs/common';
 import { ChannelsService } from './channels.service.js';
 
 // ApiKeyGuard attaches the introspected Center user id onto the request.
@@ -32,11 +32,33 @@ export class ChannelsController {
         bypassUserIds: Array.isArray(body.bypassUserIds)
           ? (body.bypassUserIds as string[])
           : [],
+        purpose: body.purpose as string | undefined,
       },
       userId,
     );
   }
 
+  // Patch a channel's free-form purpose. Body: { purpose: string }. Pass
+  // empty string to clear. Auth: channel member (or anyone for public
+  // channels, mirroring close()). Frontend doesn't call this today —
+  // intended for agent-side use (fabric-channel-set-purpose tool).
+  @Patch(':id')
+  patch(
+    @Req() req: AuthedRequest,
+    @Param('id') channelId: string,
+    @Body() body: Record<string, unknown>,
+  ) {
+    const userId = req.userId ?? '';
+    if (!userId) throw new UnauthorizedException('missing user');
+    // Only `purpose` is patchable today. Future patchable fields would
+    // get their own typed branch; we explicitly NOT allow {} no-op patches
+    // because that signals a caller bug.
+    if (typeof body.purpose !== 'string') {
+      throw new BadRequestException('purpose (string) is required');
+    }
+    return this.channelsService.updatePurpose(channelId, userId, body.purpose);
+  }
+
   // Move an order member into the bypass list (discuss/work only).
   @Post(':id/bypass')
   bypass(

src/channels/channels.service.ts

@@ -5,8 +5,9 @@ import { Channel } from '../entities/channel.entity.js';
 import { ChannelMember } from '../entities/channel-member.entity.js';
 import { WakeMapping } from '../entities/wake-mapping.entity.js';
 import { TurnService } from './turn.service.js';
+import { RealtimeGateway } from '../realtime/realtime.gateway.js';
 
-const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom'] as const;
+const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm', 'announce'] as const;
 type XType = (typeof X_TYPES)[number];
 
 type CreateChannelInput = {
@@ -23,6 +24,10 @@ type CreateChannelInput = {
   // discuss/work only: members excluded from rotation (no wakeup unless
   // @-mentioned). order and bypass partition the members disjointly.
   bypassUserIds?: string[];
+  // Free-form description of what this channel is for. Optional; agents
+  // typically fill it when creating, members can later edit via
+  // PATCH /api/channels/:id.
+  purpose?: string;
 };
 
 @Injectable()
@@ -35,8 +40,34 @@ export class ChannelsService {
     @InjectRepository(WakeMapping)
     private readonly wakeRepo: Repository<WakeMapping>,
     private readonly turnService: TurnService,
+    // RealtimeGateway is provided by the global RealtimeModule. Used to
+    // push channel.joined / channel.left so connected clients (e.g. the
+    // OpenClaw fabric plugin) can sub/unsub socket.io rooms immediately
+    // instead of waiting for the polling fallback.
+    private readonly realtime: RealtimeGateway,
   ) {}
 
+  // Push a channel membership change to each affected user's socket-room.
+  // Best-effort: offline users see the new state on their next connect
+  // (the inbound runs an initial channel-list fetch on connect).
+  private notifyMembership(
+    kind: 'joined' | 'left',
+    channelId: string,
+    userIds: string[] | Set<string>,
+    extra: Record<string, unknown> = {},
+  ): void {
+    const ids = userIds instanceof Set ? [...userIds] : userIds;
+    const payload = {
+      channelId,
+      ...extra,
+      occurredAt: new Date().toISOString(),
+    };
+    for (const u of ids) {
+      if (!u) continue;
+      this.realtime.emitToUser(u, `channel.${kind}`, { ...payload, userId: u });
+    }
+  }
+
   // Channels visible to a user within a guild:
   //  - every public channel of the guild (incl. ones created before the user
   //    joined the guild), OR
@@ -93,6 +124,7 @@ export class ChannelsService {
       if (channel.xType === 'discuss' || channel.xType === 'work') {
         await this.turnService.onMemberAdded(channelId, userId);
       }
+      this.notifyMembership('joined', channelId, [userId], { xType: channel.xType });
     }
     return { status: 'ok', channelId, userId, member: true };
   }
@@ -102,11 +134,14 @@ export class ChannelsService {
     if (!channel) throw new NotFoundException('channel not found');
 
     // remove every channel-scoped row that references this user
-    await this.memberRepo.delete({ channelId, userId });
+    const deleted = await this.memberRepo.delete({ channelId, userId });
     await this.wakeRepo.delete({ channelId, userId });
     if (channel.xType === 'discuss' || channel.xType === 'work') {
       await this.turnService.onMemberRemoved(channelId, userId);
     }
+    if ((deleted.affected ?? 0) > 0) {
+      this.notifyMembership('left', channelId, [userId], { xType: channel.xType });
+    }
     return { status: 'ok', channelId, userId, member: false };
   }
 
@@ -130,14 +165,23 @@ export class ChannelsService {
       .map((x) => String(x ?? '').trim())
       .filter(Boolean);
 
+    // dm channels are always private (a 1:1 conversation); never public.
+    // dm is not unique — multiple dm channels between the same users are
+    // allowed (create() always makes a fresh one, no dedup).
+    const isPublic = xType === 'dm' ? false : Boolean(input.isPublic);
+
+    const purposeRaw = String(input.purpose ?? '').trim();
+    const purpose = purposeRaw === '' ? null : purposeRaw;
+
     const channel = await this.channelRepo.save(
       this.channelRepo.create({
         guildId,
         name,
         xType,
         kind: input.kind === 'announcement' ? 'announcement' : 'text',
-        isPrivate: !input.isPublic,
-        isPublic: Boolean(input.isPublic),
+        isPrivate: !isPublic,
+        isPublic,
+        purpose,
         lastSeq: 0,
       }),
     );
@@ -155,6 +199,12 @@ export class ChannelsService {
       [...memberIds].map((userId) => this.memberRepo.create({ channelId: channel.id, userId })),
     );
 
+    // Push channel.joined to every seeded member (creator + invitees +
+    // triage on-duty) so their connected sockets sub the new room
+    // immediately. Skips offline users — next connect's channel-list
+    // fetch covers them.
+    this.notifyMembership('joined', channel.id, memberIds, { xType });
+
     // wake_mapping: triage -> the on-duty user; custom -> each listener
     const wakeUserIds = new Set<string>();
     if (xType === 'triage') wakeUserIds.add(onDuty);
@@ -177,6 +227,27 @@ export class ChannelsService {
     return channel;
   }
 
+  // Update a channel's free-form purpose. Any channel member may do this
+  // (or any guild user if the channel is public, mirroring closeChannel's
+  // member-or-public rule). Pass an empty string to clear.
+  async updatePurpose(channelId: string, actorUserId: string, purpose: string) {
+    const channel = await this.channelRepo.findOne({ where: { id: channelId } });
+    if (!channel) throw new NotFoundException('channel not found');
+    const member = await this.memberRepo.findOne({ where: { channelId, userId: actorUserId } });
+    if (!member && !channel.isPublic) {
+      throw new ForbiddenException('not a channel member');
+    }
+    const trimmed = String(purpose ?? '').trim();
+    channel.purpose = trimmed === '' ? null : trimmed;
+    const saved = await this.channelRepo.save(channel);
+    return {
+      id: saved.id,
+      name: saved.name,
+      xType: saved.xType,
+      purpose: saved.purpose,
+    };
+  }
+
   // Move an order member into the bypass list (discuss/work only).
   // Any channel member may do this.
   async moveToBypass(channelId: string, actorUserId: string, targetUserId: string) {

src/cli/admin-refresh.ts

@@ -0,0 +1,39 @@
+// Operator convenience: force-refresh the in-memory Center admin cache
+// without waiting for the 1-day TTL. Used after `center user set-admin`
+// to make new admin visible immediately to triage delivery.
+//
+// Usage (inside the deployed container):
+//   docker exec fabric-backend-guild node dist/cli/admin-refresh.js
+//
+// Prints the (possibly null) result as JSON. Exit 0 always — a "no
+// admin" outcome is a valid state, not an error.
+
+import 'reflect-metadata';
+import { NestFactory } from '@nestjs/core';
+import { AppModule } from '../app.module.js';
+import { AdminCacheService } from '../common/admin-cache.service.js';
+
+async function main() {
+  const app = await NestFactory.createApplicationContext(AppModule, { logger: ['error', 'warn'] });
+  try {
+    const cache = app.get(AdminCacheService);
+    const before = cache.snapshot();
+    const after = await cache.get(true);
+    process.stdout.write(
+      JSON.stringify({
+        ok: true,
+        before,
+        after,
+        changed: JSON.stringify(before) !== JSON.stringify(after),
+      }) + '\n',
+    );
+  } finally {
+    await app.close();
+  }
+}
+
+void main().catch((error: unknown) => {
+  const message = error instanceof Error ? error.message : 'unknown error';
+  process.stderr.write(JSON.stringify({ ok: false, error: message }) + '\n');
+  process.exit(1);
+});

src/cli/print-commands-sync-key.ts

@@ -0,0 +1,37 @@
+// Operator convenience (Guild C-2): print the commands-sync key that this
+// guild process actually has in its environment, so it can be copied into
+// the OpenClaw plugin's FABRIC_COMMANDS_SYNC_KEY.
+//
+// Usage (inside the deployed container — authoritative, reflects compose):
+//   docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js
+//   docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js --export
+//
+// Default: prints the raw value only (so KEY=$(... ) works).
+// --export: prints `FABRIC_COMMANDS_SYNC_KEY=<value>` for pasting.
+// Exit 1 (no stdout) when unset — guild is then in the weaker
+// "any authenticated user" fallback for PUT /commands.
+
+const args = new Set(process.argv.slice(2));
+
+if (args.has('--help') || args.has('-h')) {
+  process.stderr.write(
+    'print-commands-sync-key: outputs FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY\n' +
+      '  (no flag)  print the raw key value\n' +
+      '  --export   print FABRIC_COMMANDS_SYNC_KEY=<value>\n',
+  );
+  process.exit(0);
+}
+
+const key = (process.env.FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY ?? '').trim();
+
+if (!key) {
+  process.stderr.write(
+    'FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY is not set — PUT /commands is in ' +
+      'the fallback mode (any authenticated user). Set it to harden (Guild C-2).\n',
+  );
+  process.exit(1);
+}
+
+process.stdout.write(
+  (args.has('--export') ? `FABRIC_COMMANDS_SYNC_KEY=${key}` : key) + '\n',
+);

src/commands/commands.controller.ts

@@ -1,29 +1,51 @@
 import {
   Body,
   Controller,
+  ForbiddenException,
   Get,
+  Headers,
   Put,
   Req,
   UnauthorizedException,
 } from '@nestjs/common';
+import { timingSafeEqual } from 'node:crypto';
 import { CommandsService } from './commands.service.js';
+import { SyncCommandsDto } from './dto.sync-commands.dto.js';
 
 type AuthedRequest = { userId?: string };
 
+function safeEqual(a: string, b: string): boolean {
+  const ab = Buffer.from(a);
+  const bb = Buffer.from(b);
+  if (ab.length !== bb.length) return false;
+  return timingSafeEqual(ab, bb);
+}
+
 @Controller('commands')
 export class CommandsController {
   constructor(private readonly commands: CommandsService) {}
 
-  // Plugin syncs the OpenClaw native-command catalog here (any authenticated
-  // agent/user; idempotent full replace).
+  // Guild C-2: catalog write is privileged. When
+  // FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY is configured (recommended in
+  // production), the caller must present a matching x-commands-sync-key
+  // header — this restricts writes to the OpenClaw plugin. When unset, we
+  // fall back to "any authenticated agent/user" (never weaker than before).
+  // The body is always strictly validated + size-capped via SyncCommandsDto.
   @Put()
   sync(
     @Req() req: AuthedRequest,
-    @Body() body: { commands?: unknown[] },
+    @Body() body: SyncCommandsDto,
+    @Headers('x-commands-sync-key') syncKey?: string,
   ) {
-    if (!req.userId) throw new UnauthorizedException('missing user');
-    const commands = Array.isArray(body?.commands) ? body.commands : [];
-    return this.commands.sync(commands);
+    const configured = process.env.FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY ?? '';
+    if (configured) {
+      if (!syncKey || !safeEqual(syncKey, configured)) {
+        throw new ForbiddenException('invalid commands sync key');
+      }
+    } else if (!req.userId) {
+      throw new UnauthorizedException('missing user');
+    }
+    return this.commands.sync(body.commands as unknown[]);
   }
 
   // Frontend reads the catalog to drive `/` autocomplete.

src/commands/dto.sync-commands.dto.ts

@@ -0,0 +1,102 @@
+import {
+  ArrayMaxSize,
+  IsArray,
+  IsBoolean,
+  IsOptional,
+  IsString,
+  MaxLength,
+  ValidateNested,
+} from 'class-validator';
+import { Type } from 'class-transformer';
+
+// Guild C-2: the slash-command catalog is guild-global and rendered by the
+// frontend `/` autocomplete. Without a strict schema + caps a single
+// authenticated caller could poison it or blow up the DB / clients.
+// The global ValidationPipe runs with { whitelist, forbidNonWhitelisted },
+// so any unknown field is rejected.
+
+class CommandChoiceDto {
+  @IsString()
+  @MaxLength(200)
+  value!: string;
+
+  @IsString()
+  @MaxLength(200)
+  label!: string;
+}
+
+class CommandArgDto {
+  @IsString()
+  @MaxLength(100)
+  name!: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(500)
+  description?: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(40)
+  type?: string;
+
+  @IsOptional()
+  @IsBoolean()
+  required?: boolean;
+
+  @IsOptional()
+  @IsBoolean()
+  captureRemaining?: boolean;
+
+  @IsOptional()
+  @IsBoolean()
+  preferAutocomplete?: boolean;
+
+  // null when there are no choices (plugin sends explicit null).
+  @IsOptional()
+  @IsArray()
+  @ArrayMaxSize(100)
+  @ValidateNested({ each: true })
+  @Type(() => CommandChoiceDto)
+  choices?: CommandChoiceDto[] | null;
+}
+
+class CommandSpecDto {
+  @IsString()
+  @MaxLength(100)
+  name!: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(100)
+  nativeName?: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(500)
+  description?: string;
+
+  @IsOptional()
+  @IsBoolean()
+  acceptsArgs?: boolean;
+
+  @IsOptional()
+  @IsArray()
+  @ArrayMaxSize(50)
+  @ValidateNested({ each: true })
+  @Type(() => CommandArgDto)
+  args?: CommandArgDto[];
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(20)
+  argsParsing?: string;
+}
+
+export class SyncCommandsDto {
+  @IsArray()
+  @ArrayMaxSize(200)
+  @ValidateNested({ each: true })
+  @Type(() => CommandSpecDto)
+  commands!: CommandSpecDto[];
+}

src/common/admin-cache.service.ts

@@ -0,0 +1,73 @@
+/**
+ * Center-scoped admin cache.
+ *
+ * Holds the at-most-one admin user (email + userId) fetched from Center.
+ * Used to decide who to deliver triage messages to as a silent observer
+ * (wake=false), regardless of on-duty / mention status.
+ *
+ * Refresh policy (per spec, 2026-05-22):
+ *   • TTL = 1 day. Center admin changes are rare; agents tolerate a
+ *     day's stale cache without surprises
+ *   • on first lookup the cache lazy-fetches
+ *   • cli `admin refresh` forces an out-of-band refresh without waiting
+ *     for TTL expiry
+ *
+ * Failure mode: a Center fetch error is treated identically to "no
+ * admin" — guild keeps operating without an observer. The cache holds
+ * the failed-fetch decision for the same TTL so we don't hammer Center.
+ */
+
+import { Injectable, Logger } from '@nestjs/common';
+import { fetchAdminEmail } from './center-auth.js';
+
+const ADMIN_CACHE_TTL_MS = 24 * 60 * 60 * 1000;
+
+export interface CachedAdmin {
+  email: string;
+  userId: string;
+}
+
+@Injectable()
+export class AdminCacheService {
+  private readonly logger = new Logger(AdminCacheService.name);
+  private cached: CachedAdmin | null = null;
+  private cachedAt = 0;
+  private inflight: Promise<CachedAdmin | null> | null = null;
+
+  /**
+   * Return the cached admin, fetching from Center if the cache is empty
+   * or older than the TTL. Returns null if no admin is set.
+   *
+   * `force=true` bypasses the cache and refreshes immediately — used by
+   * the cli refresh command.
+   */
+  async get(force = false): Promise<CachedAdmin | null> {
+    const fresh = Date.now() - this.cachedAt < ADMIN_CACHE_TTL_MS;
+    if (!force && this.cachedAt > 0 && fresh) {
+      return this.cached;
+    }
+    if (this.inflight) return this.inflight;
+
+    this.inflight = (async () => {
+      try {
+        const result = await fetchAdminEmail();
+        this.cached = result;
+        this.cachedAt = Date.now();
+        this.logger.log(
+          `admin cache refreshed: ${result ? `${result.email} (${result.userId})` : 'no admin set'}`,
+        );
+        return result;
+      } finally {
+        this.inflight = null;
+      }
+    })();
+    return this.inflight;
+  }
+
+  /** Snapshot of the cached admin (no fetch). Returns null if not yet
+   *  populated. Used by the hot delivery path which doesn't want to
+   *  block on a Center round-trip. */
+  snapshot(): CachedAdmin | null {
+    return this.cached;
+  }
+}

src/common/center-auth.ts

@@ -26,6 +26,31 @@ export async function introspectGuildToken(token: string): Promise<{ active: boo
   };
 }
 
+/**
+ * Fetch the single Center-scoped admin user (if any).
+ * Same x-api-key auth as introspect / resolve-names.
+ * Returns `null` when no admin is set OR the request fails (treated
+ * identically — the guild simply falls back to "no admin observer").
+ */
+export async function fetchAdminEmail(): Promise<{ email: string; userId: string } | null> {
+  const centerBaseUrl = process.env.FABRIC_BACKEND_GUILD_CENTER_BASE_URL;
+  const centerApiKey = process.env.FABRIC_BACKEND_GUILD_CENTER_API_KEY;
+  if (!centerBaseUrl || !centerApiKey) return null;
+
+  try {
+    const res = await fetch(`${centerBaseUrl}/api/auth/admin-email`, {
+      method: 'GET',
+      headers: { 'x-api-key': centerApiKey },
+    });
+    if (!res.ok) return null;
+    const data = (await res.json()) as { email?: string; userId?: string } | null;
+    if (!data || !data.email || !data.userId) return null;
+    return { email: data.email, userId: data.userId };
+  } catch {
+    return null;
+  }
+}
+
 // Resolve <@user.name:NAME> names to userIds within this guild node via
 // Center. Unresolved names are simply absent from the returned map.
 export async function resolveUserNames(names: string[]): Promise<Record<string, string>> {

src/database.config.ts

@@ -14,6 +14,7 @@ import { IdempotencyRecord } from './entities/idempotency-record.entity.js';
 import { StoredFile } from './entities/stored-file.entity.js';
 import { ChannelCanvas } from './entities/channel-canvas.entity.js';
 import { GuildCommand } from './entities/guild-command.entity.js';
+import { AgentPresence } from './entities/agent-presence.entity.js';
 
 export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
   type: 'mysql',
@@ -38,6 +39,7 @@ export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
     StoredFile,
     ChannelCanvas,
     GuildCommand,
+    AgentPresence,
   ],
   synchronize: (process.env.FABRIC_BACKEND_GUILD_DB_SYNC ?? 'true') === 'true',
   logging: (process.env.FABRIC_BACKEND_GUILD_DB_LOGGING ?? 'false') === 'true',

src/entities/agent-presence.entity.ts

@@ -0,0 +1,35 @@
+import { Column, Entity, PrimaryColumn, UpdateDateColumn } from 'typeorm';
+
+/**
+ * Per-user (typically agent) presence cache.
+ *
+ * Populated by Fabric.OpenclawPlugin's presence-sync loop: every ~30s
+ * it reads each connected agent's HF status from the cross-plugin
+ * `globalThis.__hfAgentStatus.get(agentId)` (exposed by
+ * HarborForge.OpenclawPlugin) and pushes diffs via
+ * `PUT /agents/:userId/presence`.
+ *
+ * Used by `RealtimeGateway.computeDelivery` for `announce`-type
+ * channels to skip delivery to recipients whose status is `busy`.
+ * Defaults to `unknown` if no row exists (treated as not-busy).
+ */
+@Entity('agent_presences')
+export class AgentPresence {
+  // Same id as the Fabric Center user id (UUID v4 string, char(36)).
+  @PrimaryColumn({ type: 'char', length: 36 })
+  userId!: string;
+
+  @Column({
+    type: 'enum',
+    enum: ['idle', 'on_call', 'busy', 'exhausted', 'offline', 'unknown'],
+    default: 'unknown',
+  })
+  status!: 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
+
+  /** Free-text source tag for debugging ("hf-plugin", "manual", etc.). */
+  @Column({ type: 'varchar', length: 64, default: 'unknown' })
+  source!: string;
+
+  @UpdateDateColumn()
+  updatedAt!: Date;
+}

src/entities/channel.entity.ts

@@ -16,13 +16,23 @@ export class Channel {
   @Column({
     name: 'x_type',
     type: 'enum',
-    enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom'],
+    enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm', 'announce'],
   })
-  xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom';
+  xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm' | 'announce';
 
   @Column({ type: 'varchar', length: 16, default: 'text' })
   kind!: 'text' | 'announcement';
 
+  // Free-form description of what this channel is for — what topics get
+  // posted, who participates, why it exists. Surfaced via GET /api/channels
+  // so agents can pick a channel by intent ("which announce channel is for
+  // debate broadcasts?") without channel id hard-coded into workflows.
+  // Any channel member can set it via PATCH /api/channels/:id (writes
+  // require membership the same way moveToBypass / close do). The frontend
+  // create form does NOT post this today — purpose stays optional.
+  @Column({ type: 'text', nullable: true })
+  purpose!: string | null;
+
   @Column({ type: 'boolean', default: false })
   isPrivate!: boolean;
 

src/messaging/messaging.controller.ts

@@ -3,6 +3,7 @@ import {
   ConflictException,
   Controller,
   Delete,
+  ForbiddenException,
   Get,
   Headers,
   NotFoundException,
@@ -10,14 +11,17 @@ import {
   Patch,
   Post,
   Query,
+  Req,
 } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { DataSource, Repository } from 'typeorm';
 import { CreateMessageDto } from './dto.create-message.dto.js';
 import { Channel } from '../entities/channel.entity.js';
+import { ChannelMember } from '../entities/channel-member.entity.js';
 import { Message } from '../entities/message.entity.js';
 import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
 import { WakeMapping } from '../entities/wake-mapping.entity.js';
+import { AdminCacheService } from '../common/admin-cache.service.js';
 import { parseSlashCommand } from '../channels/slash-commands.js';
 import { parseMentions, extractNameMentions, replaceNameMentions } from '../channels/mentions.js';
 import { resolveUserNames } from '../common/center-auth.js';
@@ -36,6 +40,8 @@ export class MessagingController {
     private readonly dataSource: DataSource,
     @InjectRepository(Channel)
     private readonly channelRepo: Repository<Channel>,
+    @InjectRepository(ChannelMember)
+    private readonly memberRepo: Repository<ChannelMember>,
     @InjectRepository(Message)
     private readonly messageRepo: Repository<Message>,
     @InjectRepository(IdempotencyRecord)
@@ -45,6 +51,7 @@ export class MessagingController {
     private readonly turn: TurnService,
     private readonly events: EventsService,
     private readonly realtime: RealtimeGateway,
+    private readonly adminCache: AdminCacheService,
   ) {}
 
   private async getIdempotentResponse(
@@ -86,6 +93,19 @@ export class MessagingController {
     };
   }
 
+  // Channel-participant gate (Guild C-1): public channels are readable/
+  // writable by any authenticated user; private channels require explicit
+  // channel_members membership. Returns the channel so callers can reuse it.
+  private async assertParticipant(channelId: string, userId: string): Promise<Channel> {
+    const channel = await this.channelRepo.findOne({ where: { id: channelId } });
+    if (!channel) throw new NotFoundException('channel not found');
+    if (channel.isPublic) return channel;
+    if (!userId) throw new ForbiddenException('not a channel member');
+    const member = await this.memberRepo.findOne({ where: { channelId, userId } });
+    if (!member) throw new ForbiddenException('not a channel member');
+    return channel;
+  }
+
   // Persists one message (allocates a seq under a channel row lock) and
   // returns its view. Used for normal messages and for guild /ack messages.
   private async persistMessage(
@@ -136,20 +156,30 @@ export class MessagingController {
   async create(
     @Param('id') channelId: string,
     @Body() body: CreateMessageDto,
+    @Req() req: { userId?: string },
     @Headers('idempotency-key') idempotencyKey?: string,
   ) {
     const scope = `POST:/channels/${channelId}/messages`;
     const existed = await this.getIdempotentResponse(scope, idempotencyKey);
     if (existed) return existed;
 
-    const channel = await this.channelRepo.findOne({ where: { id: channelId } });
-    if (!channel) throw new NotFoundException('channel not found');
+    // Guild C-1: caller must be a participant of the channel, and the
+    // author is always the authenticated user — body.authorUserId is
+    // ignored so a caller can never post as someone else.
+    //
+    // announce channels: any participant can POST. Use case is one-off
+    // recruitment / broadcast messages posted by the agent that just
+    // created the originating topic (e.g. dialectic invites). No
+    // server-side privileged path — author is always a real user.
+    const userId = String(req.userId ?? '');
+    if (!userId) throw new ForbiddenException('missing user');
+    const channel = await this.assertParticipant(channelId, userId);
     if (channel.closed) {
       throw new ConflictException({ error: 'channel_closed', message: 'channel is closed' });
     }
     const xType = channel.xType ?? 'general';
     const isRotating = xType === 'discuss' || xType === 'work';
-    const authorUserId = String(body.authorUserId ?? 'anonymous');
+    const authorUserId = userId;
 
     // ---- translate <@user.name:NAME> -> <@userId> (outside backticks) via
     // Center before anything else persists/parses the content
@@ -202,16 +232,19 @@ export class MessagingController {
       const decision = await this.turn.onNormalMessage(channelId, authorUserId, mentionIds);
       await this.realtime.emitMessageTargeted(channelId, responseBody, decision.wakeupUserId);
     } else {
-      // general/report/triage/custom: wakeup from x_type + wake_mapping;
-      // general also honors the message's at-list
+      // general/report/triage/custom: 3-state delivery
+      // (wake / observer / skip) — see realtime.gateway.computeDelivery.
+      // Center-scoped admin (cached, 1d TTL) gets `observer` on triage.
       const wakeRows = await this.wakeRepo.find({ where: { channelId } });
       const wakeUserIds = new Set(wakeRows.map((w) => w.userId));
       const mentionUserIds = new Set(mentionIds.filter((id) => id !== authorUserId));
+      const admin = await this.adminCache.get();
       await this.realtime.emitMessageCreated(channelId, responseBody, {
         xType,
         authorUserId,
         wakeUserIds,
         mentionUserIds,
+        adminUserId: admin?.userId ?? null,
       });
     }
 
@@ -223,14 +256,23 @@ export class MessagingController {
     @Param('id') channelId: string,
     @Param('messageId') messageId: string,
     @Body() body: { content?: string },
+    @Req() req: { userId?: string },
     @Headers('idempotency-key') idempotencyKey?: string,
   ) {
     const scope = `PATCH:/channels/${channelId}/messages/${messageId}`;
     const existed = await this.getIdempotentResponse(scope, idempotencyKey);
     if (existed) return existed;
 
+    // Guild C-1: participant + author-ownership.
+    const userId = String(req.userId ?? '');
+    if (!userId) throw new ForbiddenException('missing user');
+    await this.assertParticipant(channelId, userId);
+
     const item = await this.messageRepo.findOne({ where: { channelId, messageId } });
     if (!item) return { status: 'not_found' };
+    if (item.authorUserId !== userId) {
+      throw new ForbiddenException('not the message author');
+    }
 
     const now = Date.now();
     const createdAt = new Date(item.createdAt).getTime();
@@ -259,14 +301,23 @@ export class MessagingController {
   async remove(
     @Param('id') channelId: string,
     @Param('messageId') messageId: string,
+    @Req() req: { userId?: string },
     @Headers('idempotency-key') idempotencyKey?: string,
   ) {
     const scope = `DELETE:/channels/${channelId}/messages/${messageId}`;
     const existed = await this.getIdempotentResponse(scope, idempotencyKey);
     if (existed) return existed;
 
+    // Guild C-1: participant + author-ownership.
+    const userId = String(req.userId ?? '');
+    if (!userId) throw new ForbiddenException('missing user');
+    await this.assertParticipant(channelId, userId);
+
     const item = await this.messageRepo.findOne({ where: { channelId, messageId } });
     if (!item) return { status: 'not_found' };
+    if (item.authorUserId !== userId) {
+      throw new ForbiddenException('not the message author');
+    }
 
     item.isDeleted = true;
     item.deletedAt = new Date();
@@ -304,10 +355,14 @@ export class MessagingController {
   @Get()
   async listBySeq(
     @Param('id') channelId: string,
+    @Req() req: { userId?: string },
     @Query('seq_from') seqFrom?: string,
     @Query('seq_to') seqTo?: string,
     @Query('limit') limit?: string,
   ) {
+    // Guild C-1: only participants may read channel history.
+    const userId = String(req.userId ?? '');
+    if (!userId) throw new ForbiddenException('missing user');
     const from = seqFrom ? Number(seqFrom) : 1;
     const to = seqTo ? Number(seqTo) : Number.MAX_SAFE_INTEGER;
     const safeLimit = clampLimit(limit, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT);
@@ -327,10 +382,7 @@ export class MessagingController {
       };
     }
 
-    const channel = await this.channelRepo.findOne({ where: { id: channelId } });
-    if (!channel) {
-      throw new NotFoundException('channel not found');
-    }
+    const channel = await this.assertParticipant(channelId, userId);
 
     const qb = this.messageRepo
       .createQueryBuilder('m')

src/messaging/messaging.module.ts

@@ -2,12 +2,16 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { MessagingController } from './messaging.controller.js';
 import { Channel } from '../entities/channel.entity.js';
+import { ChannelMember } from '../entities/channel-member.entity.js';
 import { Message } from '../entities/message.entity.js';
 import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
 import { WakeMapping } from '../entities/wake-mapping.entity.js';
+import { AdminCacheService } from '../common/admin-cache.service.js';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Channel, Message, IdempotencyRecord, WakeMapping])],
+  imports: [TypeOrmModule.forFeature([Channel, ChannelMember, Message, IdempotencyRecord, WakeMapping])],
   controllers: [MessagingController],
+  providers: [AdminCacheService],
+  exports: [AdminCacheService],
 })
 export class MessagingModule {}

src/realtime/realtime.gateway.ts

@@ -11,17 +11,94 @@ import { Logger } from '@nestjs/common';
 import { Server, Socket } from 'socket.io';
 import { introspectGuildToken } from '../common/center-auth.js';
 
-type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom';
+type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm' | 'announce';
 
-// Wakeup for non-rotating channels only (general/report/triage/custom).
-// discuss/work go through TurnService + emitMessageTargeted, never here.
-// Precedence:
-//  1. the author never gets woken by their own message
-//  2. triage/custom: only wake users in the channel's wake_mapping
-//     (mentions change nothing here)
-//  3. general: if the message has an at-list, wake only the at'd users;
-//     otherwise wake everyone
-//  4. report (and anything else): wake nobody
+/**
+ * Cross-presence info needed by `announce`-type delivery: a recipient
+ * with hf-side status === 'busy' has the message discarded silently
+ * (don't enter their session, no UI emit). Other statuses + non-announce
+ * channels are unaffected. Presence is sourced from the
+ * `agent_presences` table populated by Fabric.OpenclawPlugin's
+ * presence-sync loop (which reads from HF plugin's `__hfAgentStatus`).
+ */
+export type PresenceStatus = 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
+
+/**
+ * Per-recipient delivery decision for a non-rotating channel message.
+ *
+ *   • `wake` — push the event AND wake the recipient (model turn fires)
+ *   • `observer` — push the event with wakeup=false (silent; UI displays
+ *     but the openclaw plugin records-only without dispatch). Currently
+ *     used for the Center admin observing triage traffic
+ *   • `skip` — don't even emit the event to this recipient
+ *
+ * Wakeup-only channels (general/report/dm/custom) never return
+ * 'observer'; the legacy behaviour is preserved end-to-end.
+ *
+ * Precedence for triage (the only place 'skip' / 'observer' fire):
+ *   1. author never gets back their own message
+ *   2. wake_mapping (on-duty) → wake
+ *   3. mention → wake (NEW: was 'skip' before — see Fabric PR 'triage
+ *      mention exception')
+ *   4. admin (Center-scoped, at most one) → observer
+ *   5. everyone else → skip (was 'deliver, wakeup=false' before)
+ */
+export type DeliveryDecision = 'wake' | 'observer' | 'skip';
+
+export interface ComputeDeliveryArgs {
+  xType: XType;
+  recipientUserId: string;
+  authorUserId: string;
+  wakeUserIds: Set<string>;
+  mentionUserIds?: Set<string>;
+  /** Single Center-scoped admin userId, or null. */
+  adminUserId?: string | null;
+  /** Recipient's current presence; only consulted for `announce` xType. Defaults to 'unknown' (treated as not-busy). */
+  recipientPresence?: PresenceStatus;
+}
+
+export function computeDelivery(args: ComputeDeliveryArgs): DeliveryDecision {
+  const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds, adminUserId, recipientPresence } = args;
+  if (recipientUserId === authorUserId) return 'skip';
+
+  switch (xType) {
+    case 'triage':
+      if (wakeUserIds.has(recipientUserId)) return 'wake';
+      if (mentionUserIds?.has(recipientUserId)) return 'wake';
+      if (adminUserId && recipientUserId === adminUserId) return 'observer';
+      return 'skip';
+    case 'general':
+      if (mentionUserIds && mentionUserIds.size > 0) {
+        return mentionUserIds.has(recipientUserId) ? 'wake' : 'observer';
+      }
+      return 'wake';
+    case 'custom':
+      // wake_mapping decides who wakes; everyone else still sees the
+      // message (observer) — preserves the legacy "deliver to all, wake
+      // some" contract for custom channels.
+      return wakeUserIds.has(recipientUserId) ? 'wake' : 'observer';
+    case 'dm':
+      return 'wake';
+    case 'announce':
+      // System-broadcast channels (e.g. Dialectic topic announcements).
+      // Recipients with HF status === 'busy' have the message discarded
+      // silently — busy agents should not be distracted by signup pings
+      // they can't act on. All other presences (idle/on_call/exhausted/
+      // offline/unknown) get the message as 'observer' (no wake): the
+      // channel itself is browsable; agents proactively decide what to
+      // do with announcements when they next look at their inbox.
+      if (recipientPresence === 'busy') return 'skip';
+      return 'observer';
+    default:
+      // report (and anything else): deliver as observer, no wake
+      return 'observer';
+  }
+}
+
+/**
+ * @deprecated Use computeDelivery (returns 3-state). Kept for any
+ * external callers; treats 'observer' and 'skip' both as `false`.
+ */
 export function computeWakeup(args: {
   xType: XType;
   recipientUserId: string;
@@ -29,20 +106,7 @@ export function computeWakeup(args: {
   wakeUserIds: Set<string>;
   mentionUserIds?: Set<string>;
 }): boolean {
-  const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds } = args;
-  if (recipientUserId === authorUserId) return false;
-  switch (xType) {
-    case 'general':
-      if (mentionUserIds && mentionUserIds.size > 0) {
-        return mentionUserIds.has(recipientUserId);
-      }
-      return true;
-    case 'triage':
-    case 'custom':
-      return wakeUserIds.has(recipientUserId);
-    default:
-      return false;
-  }
+  return computeDelivery(args) === 'wake';
 }
 
 @WebSocketGateway({
@@ -58,6 +122,12 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
   private readonly logger = new Logger(RealtimeGateway.name);
   private readonly onlineUsers = new Set<string>();
 
+  // Optional: injected at module wiring time. Used by emitMessageCreated
+  // to pre-load recipient presence for announce-type channels.
+  // Typed loosely to avoid a circular import between realtime and agents
+  // modules; the actual interface lives in agents/agent-presence.service.
+  presence?: { getStatusMap(ids: string[]): Promise<Map<string, PresenceStatus>> };
+
   private userIdFromClient(client: Socket): string {
     const authUser = client.handshake.auth?.userId;
     const headerUser = client.handshake.headers['x-user-id'];
@@ -93,6 +163,10 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
     const userId = result.user.id || this.userIdFromClient(client);
     client.data.userId = userId;
     this.onlineUsers.add(userId);
+    // Per-user room: lets server code emit user-scoped events (e.g.
+    // channel.joined when membership changes) without bookkeeping a
+    // userId→sockets map. All of this user's sockets receive the event.
+    client.join(`user:${userId}`);
     this.server.emit('presence.online', {
       userId,
       onlineCount: this.onlineUsers.size,
@@ -168,7 +242,19 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
     this.server.to(`channel:${channelId}`).emit(event, data);
   }
 
-  // Emits message.created per-recipient so each carries its own `wakeup` flag.
+  // Emit a user-scoped event to all sockets currently connected for `userId`
+  // (via the `user:<userId>` room joined in handleConnection). No-op for
+  // offline users — the next connect's initial channel-list fetch covers it.
+  emitToUser(userId: string, event: string, data: Record<string, unknown>): void {
+    if (!userId) return;
+    this.server.to(`user:${userId}`).emit(event, data);
+  }
+
+  // Emits message.created per-recipient using the 3-state delivery
+  // decision (wake / observer / skip). Skipped recipients receive
+  // nothing — used by triage channels to keep non-on-duty / non-mention
+  // / non-admin users completely out of the loop, and by announce
+  // channels to suppress delivery to recipients whose presence is busy.
   async emitMessageCreated(
     channelId: string,
     data: Record<string, unknown>,
@@ -177,19 +263,41 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
       authorUserId: string;
       wakeUserIds: Set<string>;
       mentionUserIds?: Set<string>;
+      /** Single Center-scoped admin userId (or null). */
+      adminUserId?: string | null;
     },
   ): Promise<void> {
     const sockets = await this.server.in(`channel:${channelId}`).fetchSockets();
+
+    // For announce-type channels, pre-load presence for all recipients
+    // in one query so the per-recipient loop doesn't fan out to N round
+    // trips. For other xTypes, presence is irrelevant — skip the lookup.
+    let presenceMap: Map<string, PresenceStatus> | undefined;
+    if (ctx.xType === 'announce' && this.presence) {
+      const recipientIds = sockets
+        .map((s) => (typeof s.data.userId === 'string' ? (s.data.userId as string) : ''))
+        .filter((id) => id && !id.startsWith('anon:'));
+      presenceMap = await this.presence.getStatusMap(recipientIds);
+    }
+
     for (const s of sockets) {
       const recipientUserId = typeof s.data.userId === 'string' ? s.data.userId : `anon:${s.id}`;
-      const wakeup = computeWakeup({
+      const decision = computeDelivery({
         xType: ctx.xType,
         recipientUserId,
         authorUserId: ctx.authorUserId,
         wakeUserIds: ctx.wakeUserIds,
         mentionUserIds: ctx.mentionUserIds,
+        adminUserId: ctx.adminUserId,
+        recipientPresence: presenceMap?.get(recipientUserId) ?? 'unknown',
+      });
+      if (decision === 'skip') continue;
+      s.emit('message.created', {
+        ...data,
+        channelId,
+        wakeup: decision === 'wake',
+        xType: ctx.xType,
       });
-      s.emit('message.created', { ...data, channelId, wakeup });
     }
   }
 

src/realtime/realtime.module.ts

@@ -1,9 +1,25 @@
-import { Global, Module } from '@nestjs/common';
+import { Global, Module, OnModuleInit } from '@nestjs/common';
 import { RealtimeGateway } from './realtime.gateway.js';
+import { AgentPresenceModule } from '../agents/agent-presence.module.js';
+import { AgentPresenceService } from '../agents/agent-presence.service.js';
 
 @Global()
 @Module({
+  imports: [AgentPresenceModule],
   providers: [RealtimeGateway],
   exports: [RealtimeGateway],
 })
-export class RealtimeModule {}
+export class RealtimeModule implements OnModuleInit {
+  // Wire presence into the gateway at startup. Using assignment (vs
+  // constructor injection) keeps the gateway free of the agents-module
+  // import — no risk of circular dependency, and announce-channel
+  // delivery degrades gracefully (presence stays undefined → 'unknown'
+  // status → no busy-discard) if AgentPresenceModule is ever removed.
+  constructor(
+    private readonly gateway: RealtimeGateway,
+    private readonly presence: AgentPresenceService,
+  ) {}
+  onModuleInit(): void {
+    this.gateway.presence = this.presence;
+  }
+}