fix(server): wake inbound queue notices from committed changes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -1737,3 +1737,19 @@ base-publication hook. The hook fixes a reproduced commit-order wake gap; its
 16 cases at `bcf2f38ff5`; final typecheck passes too. Remote checks/review are pending.
 Parent #17606 now has passing checks and Greptile 5/5 on `ed76256b97`; all three
 review threads are resolved. This is not completion of the broader stack.
+
+## Current gate update: native progress and inbound notices (2026-10-25)
+
+#17639 at `6c40618ec1bb098e1c7fe4098133d1de71d287ee` now has all checks
+passing (CI run 40218349023), a fresh Greptile 5/5 review of that commit, and
+zero review threads. The circular signal-payload and over-wide atomic-event
+selection failures were repaired, not bypassed by rerunning unchanged failures.
+The PR remains unmerged.
+
+Slice 30 now registers its scoped notice worker and removes normal inbound
+notices from periodic publication maintenance. Focused acceptance passes 42
+cases. Its full verification and new PR are still pending. See the dated inbound
+wake notice audit for the producer inventory, regressions and fixture corrections.
+This update does not clear the older Grok/runtime-discovery CI failures, voice
+session recovery review, or inbound voice cleanup ownership findings, and it does
+not shrink the remaining original checklist.

doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md

@@ -220,3 +220,72 @@ fresh `paperclip_sleep30_full_chat_20261025_test` database, with output at
 `/tmp/sleep30-full-chat-acceptance.log`; its result must be read before claiming
 acceptance. Typecheck after merging the second parent repair passed in
 `/tmp/sleep30-after-parent-typecheck.log`.
+
+## Producer and lifecycle completion (2026-10-25)
+
+The full chat suite at `c579148977` finished successfully: 1,458 tests in
+`/tmp/sleep30-full-chat-acceptance.log`. This was the baseline before the
+cancellation and privacy changes below, not acceptance of later edits.
+
+Real service regressions reproduced stale visible queue notices after agent
+cancellation, budget cancellation, and company archival. Signals now belong to
+those owners' transactions, as well as Slack Stop and deferred queue cancellation
+and failure. Agent cancellation now rechecks pending status and absence of a run
+at update time, so it cannot cancel a receipt promoted after candidate selection.
+Caller-owned transactions preserve outer commit and rollback. Company lifecycle
+intentionally owns its root transaction; its test respects that contract.
+
+Task status/assignment changes and privacy changes can restore eligibility.
+Task privacy can affect descendants, so it invalidates the company scope. Explicit
+task grants and private-project membership use their existing persistence owners;
+the project-member helper owns a small transaction that also nests under callers.
+Project visibility changes use the existing project transaction. Shared chat
+identity, endpoint, resource and membership signals already cover the other
+current-authority predicates. Ordinary health refreshes and receipt cursor touches
+do not signal this worker. Run-bound settlement and Dot/Muse consumption do not
+create a not-started notice; their existing response/progress owners retain that
+policy. Failed-run retry notices remain a separate owner.
+
+Failed inbound-action settlement and repaired delivery acceptance now invalidate
+normal notices when a valid durable request exists. Preparing actions whose
+underlying deliveries were rejected cannot satisfy the notice predicate and do
+not acquire a notice obligation. Publication completion already signals the task
+inside the transaction that saves the provider receipt and message link. A new
+race test holds provider I/O, removes the source, observes the notice worker
+settle before the receipt exists, and proves that the eventual receipt commit
+reactivates the removed-source edit.
+
+The app now registers `chatInboundWakeNoticeWork` with the existing coordinator.
+`processPublicationMaintenance` retains only the separate failed-run retry notice
+pass, plus its other unmigrated duties. The explicit/manual combined
+`enqueueInboundWakeupPublications` API remains available.
+
+Focused acceptance at the current source passes 42 cases in
+`/tmp/sleep30-wired-acceptance.log`. These include actual admission after idle,
+scoped/manual cursor pagination, denied history, receipt-lock contention,
+transaction rollback and commit, queue rejection/failure, agent/budget/company
+cancellation, privacy restoration through task grants and project membership,
+signed Slack Stop, late provider receipt, restart deduplication, writes after
+shutdown, first write and failed selection during drain, and a commit arriving
+while an empty selection returns. Quiet assertions observe root SQL entry points
+after advancing fake time by two minutes. The lifecycle tests do not prove host
+or database suspension, and no Cloud sleep-safety blocker is relaxed.
+
+Preserved failures and fixture corrections:
+- `/tmp/sleep30-cancellation-before.log`: agent cancellation missed its wake;
+  an attempted agent reassignment fixture was rejected by the existing external
+  channel binding rule. The unsupported fixture was removed, not bypassed.
+- `/tmp/sleep30-budget-company-before.log`: budget cancellation missed its wake;
+  company lifecycle correctly rejected a caller transaction. The root-company
+  reproduction in `/tmp/sleep30-company-root-before.log` then showed its missing
+  wake. All five initial retirement cases passed after the repair.
+- `/tmp/sleep30-notice-privacy-before.log`: opening a private task did not wake
+  its saved notice. The repair passes. Initial grant tests used an unlinked
+  external principal as a user; corrected fixtures establish a linked board user
+  through the normal chat setup before admission.
+- The first signed Stop fixture had an invalid Slack channel ID and a verifying,
+  rather than qualified active, endpoint. Correcting both makes the signed event
+  reach cancellation. `/tmp/sleep30-signed-stop-qualified.log` passes.
+
+Workspace typecheck/build, full acceptance after startup wiring, PR publication,
+and latest-head CI/review are still required before this slice is ready.

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -79,6 +79,7 @@ describe("createChatReconciliationCoordinator", () => {
     expect(maintenance).not.toContain("chatChannels.voice.");
     expect(maintenance).not.toContain("enqueueSafeNativeChatProgress");
     expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,");
+    expect(source).toContain("chatInboundWakeNoticeWork(chatChannels).register(deliveryWork,");
     expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
     expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,");
     const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8");
@@ -87,6 +88,10 @@ describe("createChatReconciliationCoordinator", () => {
     expect(teams).not.toContain("selectDistinct");
     const chat = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8");
     expect(chat).not.toContain("scheduleTeamsFileMaintenance");
+    const publicationMaintenance = chat.slice(chat.indexOf("async function processPublicationMaintenance("),
+      chat.indexOf("async function schedulePendingPublications("));
+    expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications(");
+    expect(publicationMaintenance).toContain("await enqueueFailedChatRetryPublications(");
   });
 
   it("separates live voice recovery from the remaining periodic maintenance", () => {

server/src/__tests__/chat-channels.integration.test.ts

@@ -178,6 +178,9 @@ import {
 } from "../services/chat-teams-personal-recipient.js";
 import * as discordQuestionForms from "../services/chat-discord-question-forms.js";
 import { issueService } from "../services/issues.js";
+import { companyService } from "../services/companies.js";
+import { ensureProjectAccessMember } from "../services/projects.js";
+import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { getExternalChannelBindingSummary } from "../services/chat-channel-binding.js";
 import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js";
@@ -71008,10 +71011,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     expect(committed.providerRuntime.slackFilePublicationAttempts).toBe(2);
   });
 
-  async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise<void>) {
+  async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise<void>, linkedBoardUser = false) {
     const fixture = await seedCompany();
     let ownerId: string | null = null;
     const configured = await configuredSlackEndpoint(fixture, {
+      linkedBoardUser,
       wakeup: async (agentId, opts) => {
         if (admissionFails)
           throw new Error("PRIVATE admission failure token=do-not-publish");
@@ -71067,8 +71071,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
       },
     });
     const thread = makeThread({
-      channelId: "C-DEFERRED-NOTICE",
-      id: "slack:C-DEFERRED-NOTICE:1999000.1",
+      channelId: "CDEFERREDNOTICE",
+      id: "slack:CDEFERREDNOTICE:1999000.1",
     });
     const send = async (id = "1999000.1") =>
       deliverMessage({
@@ -71078,6 +71082,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         message: makeMessage({
           id,
           text: "@maya exact queued request",
+          userId: linkedBoardUser ? "UBOARD" : undefined,
           mentioned: id === "1999000.1",
         }),
         trigger: id === "1999000.1" ? "mention" : "subscribed_message",
@@ -71170,8 +71175,158 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     }
   });
 
-  it.each(["tree cancellation", "source removal"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => {
+  it("reconstructs inbound admission committed after the notice worker stops", async () => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let service: ChatChannelService | undefined;
+    const errors: unknown[] = [];
+    try {
+      const context = await deferredChatQueueFixture(false, async channel => {
+        service = channel;
+        vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+          canReconcile: () => true, onError: error => errors.push(error) });
+        await chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready;
+        await coordinator.stop();
+      });
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+        canReconcile: () => true, onError: error => errors.push(error) });
+      await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(errors).toEqual([]);
+    } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); }
+  });
+
+  it.each(["task visibility", "project membership", "task grant"] as const)("wakes a saved inbound notice when %s restores privacy access", async kind => {
+    const context = await deferredChatQueueFixture(false, undefined, true);
+    await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.conversation.issueId,
+      createdByUserId: null, responsibleUserId: null, assigneeUserId: null,
+    }).where(eq(issues.id, context.conversation.issueId));
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const settle = async () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+    }, { timeout: 10_000 });
+    try {
+      await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
+      await settle();
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      if (kind === "task visibility") {
+        await issueService(db).update(context.conversation.issueId, { visibility: "open" });
+      } else if (kind === "project membership") {
+        const [project] = await db.insert(projects).values({ companyId: context.fixture.companyId,
+          name: "Private notice project", visibility: "private",
+        }).returning();
+        await db.update(issues).set({ projectId: project.id }).where(eq(issues.id, context.conversation.issueId));
+        await ensureProjectAccessMember(db, { companyId: context.fixture.companyId, projectId: project.id,
+          subjectType: "user", subjectId: String(context.action.payload.requestedByActorId),
+        });
+      } else {
+        const app = express();
+        app.use(express.json());
+        app.use((req, _res, next) => { req.actor = { ...boardActor(context.fixture.companyId), isInstanceAdmin: true }; next(); });
+        app.use("/api", issueRoutes(db, createStorageService().storage)); app.use(errorHandler);
+        await request(app).post(`/api/issues/${context.conversation.issueId}/access-grants`).send({
+          subjectType: "user", subjectId: String(context.action.payload.requestedByActorId),
+        }).expect(201);
+      }
+      await settle();
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({
+        payload: { text: "Your follow-up is queued.", progressState: "queued" },
+      })]);
+      expect(errors).toEqual([]);
+    } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it.each(["startup", "first admission", "failed selection"] as const)("retains inbound notices through %s during drain without passive SQL", async kind => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let service: ChatChannelService | undefined;
+    let canRun = kind !== "startup";
+    const errors: unknown[] = [];
+    try {
+      const context = await deferredChatQueueFixture(false, async channel => {
+        service = channel;
+        vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+          canReconcile: () => true, onError: error => errors.push(error) });
+        const selection = kind === "failed selection"
+          ? vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockRejectedValueOnce(new Error("temporary notice selection outage"))
+          : undefined;
+        try { await chatInboundWakeNoticeWork(channel).register(coordinator, () => canRun).ready; }
+        finally { selection?.mockRestore(); }
+        canRun = false;
+      });
+      await vi.advanceTimersByTimeAsync(30_000);
+      await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+      canRun = true;
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(errors).toHaveLength(kind === "failed selection" ? 1 : 0);
+    } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); }
+  });
+
+  it("retains admission committed while an empty inbound-notice selection is returning", async () => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let service: ChatChannelService | undefined;
+    let selection: ReturnType<typeof vi.spyOn> | undefined;
+    let ready: Promise<unknown> | undefined;
+    let held = false, release!: () => void;
+    const pending = new Promise<void>(resolve => { release = resolve; });
+    const errors: unknown[] = [];
+    try {
+      const context = await deferredChatQueueFixture(false, async channel => {
+        service = channel;
+        vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+        const load = channel.loadInboundWakeNoticeScopes.bind(channel);
+        selection = vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockImplementationOnce(async (...args) => {
+          const rows = await load(...args); held = true; await pending; return rows;
+        });
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+          canReconcile: () => true, onError: error => errors.push(error) });
+        ready = chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready;
+        await vi.waitFor(() => expect(held).toBe(true));
+      });
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      release(); await ready;
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(errors).toEqual([]);
+    } finally { release(); await ready; selection?.mockRestore(); await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); }
+  });
+
+  it.each(["tree cancellation", "source removal", "agent cancellation", "budget cancellation", "company archive", "queue rejection", "queue failure"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => {
     const context = await deferredChatQueueFixture();
+    const releaseRunId = randomUUID();
+    if (change === "queue rejection" || change === "queue failure") {
+      await db.insert(heartbeatRuns).values({ id: releaseRunId, companyId: context.fixture.companyId,
+        agentId: context.fixture.assignedAgentId, status: "succeeded", finishedAt: new Date(),
+        contextSnapshot: { issueId: context.conversation.issueId },
+      });
+    }
     const errors: unknown[] = [];
     const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
       canReconcile: () => true, onError: error => errors.push(error) });
@@ -71183,26 +71338,51 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const notices = () => readChatPublicationsByEndpointId(context.endpoint.id);
     const expectedText = change === "source removal" ? "This queued message was removed."
       : "This follow-up was not started. Open the task in Paperclip for details.";
-    const retire = (database: typeof db) => change === "source removal"
-      ? issueService(database).tombstoneComment(String(context.action.payload.commentId), {
+    const retire = async (database: typeof db) => {
+      if (change === "source removal") return issueService(database).tombstoneComment(String(context.action.payload.commentId), {
         actorType: "user", userId: String(context.action.payload.requestedByActorId),
-      })
-      : issueTreeControlService(database).cancelUnclaimedWakeupsForTree(
+      });
+      if (change === "agent cancellation") return heartbeatService(database).cancelInvocationsForAgents(
+        [context.fixture.assignedAgentId], "PRIVATE cancellation",
+      );
+      if (change === "budget cancellation") return heartbeatService(database).cancelBudgetScopeWork({
+        companyId: context.fixture.companyId, scopeType: "company", scopeId: context.fixture.companyId,
+      });
+      if (change === "company archive") return companyService(database).update(context.fixture.companyId, { status: "archived" });
+      if (change === "queue rejection" || change === "queue failure") {
+        const adapter = createPostgresWakeQueueAdapter(database, {
+          resolveResponsibleUserId: async () => "owner-user",
+          getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
+          resolveSessionBeforeForWakeup: async () => null,
+        });
+        return adapter.withIssueExecutionLock({ companyId: context.fixture.companyId, runId: releaseRunId, now: new Date() }, async (_locked, ports) => {
+          const target = { companyId: context.fixture.companyId, wakeId: context.action.id, now: new Date() };
+          if (change === "queue rejection") await ports.transaction.cancelDeferredWake({ ...target, reason: "PRIVATE rejection" });
+          else await ports.transaction.failDeferredWake(target);
+          return { outcome: { kind: "released" }, postCommitEffects: [] };
+        });
+      }
+      return issueTreeControlService(database).cancelUnclaimedWakeupsForTree(
         context.fixture.companyId, context.conversation.issueId, "PRIVATE cancellation",
       );
+    };
     try {
       await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
       await settle();
       await context.service.processPendingPublications(100, { maintenance: false });
       await settle();
       expect(await notices()).toEqual([expect.objectContaining({ state: "published" })]);
-      await expect(db.transaction(async tx => {
-        await retire(tx as unknown as typeof db);
-        await vi.advanceTimersByTimeAsync(1000);
-        expect(await notices()).toHaveLength(1);
-        throw new Error("outer cancellation rollback");
-      })).rejects.toThrow("outer cancellation rollback");
-      await settle(); expect(await notices()).toHaveLength(1);
+      // Company lifecycle owns its root transaction and intentionally rejects
+      // caller transactions. The other service producers support nesting.
+      if (change !== "company archive") {
+        await expect(db.transaction(async tx => {
+          await retire(tx as unknown as typeof db);
+          await vi.advanceTimersByTimeAsync(1000);
+          expect(await notices()).toHaveLength(1);
+          throw new Error("outer cancellation rollback");
+        })).rejects.toThrow("outer cancellation rollback");
+        await settle(); expect(await notices()).toHaveLength(1);
+      }
       await retire(db);
       await settle();
       expect(await notices()).toEqual(expect.arrayContaining([expect.objectContaining({
@@ -71223,6 +71403,81 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
   });
 
+  it("rechecks source retirement after a queued notice's late provider receipt commits", async () => {
+    const context = await deferredChatQueueFixture();
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    let held = false, release!: () => void;
+    const pending = new Promise<void>(resolve => { release = resolve; });
+    let publishing: Promise<number> | undefined;
+    const provider = context.runtime.endpoints.get(context.endpoint.id)!;
+    const settle = async () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+    }, { timeout: 10_000 });
+    try {
+      await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
+      await settle();
+      provider.postHook = async () => { held = true; await pending; };
+      publishing = context.service.processPendingPublications(100, { maintenance: false });
+      await vi.waitFor(() => expect(held).toBe(true));
+      await issueService(db).tombstoneComment(String(context.action.payload.commentId), {
+        actorType: "user", userId: "owner-user",
+      });
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(coordinator.nextWakeAt()).toBeNull();
+      });
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+      release(); await publishing; provider.postHook = undefined;
+      await settle();
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({
+        payload: { text: "This queued message was removed.", progressState: "failed" },
+      })]));
+      await context.service.processPendingPublications(100, { maintenance: false });
+      await settle();
+      expect(provider.edits).toEqual([expect.objectContaining({ text: "This queued message was removed." })]);
+      expect(errors).toEqual([]);
+    } finally { release(); await publishing; provider.postHook = undefined; await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("wakes inbound notice retirement from a signed Slack Stop without a maintenance scan", async () => {
+    const context = await deferredChatQueueFixture(false, undefined, true);
+    await qualifySetupRoundTrip(context.service, context.endpoint.id, "UBOARD");
+    await context.service.test(context.endpoint.id, "owner-user");
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const settle = async () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+    }, { timeout: 10_000 });
+    try {
+      await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready;
+      await settle(); await context.service.processPendingPublications(100, { maintenance: false }); await settle();
+      const body = JSON.stringify({ type: "event_callback", team_id: "T-PAPERCLIP", event_id: "EvSleepQueueStop",
+        event: { type: "agent_session_stopped", channel: "CDEFERREDNOTICE", thread_ts: "1999000.1",
+          event_ts: `${Math.floor(Date.now() / 1000)}.999999`, streaming_message_ts: [], user: "UBOARD" },
+      });
+      const response = await context.service.handleWebhook(context.endpoint.publicId, "slack", signedSlackWebhookRequest({
+        body, contentType: "application/json", url: `https://paperclip.example/api/chat-webhooks/${context.endpoint.publicId}/slack`,
+      }));
+      expect(response.status).toBe(202);
+      await settle();
+      expect(await db.select({ status: chatActions.status, result: chatActions.result }).from(chatActions)
+        .where(and(eq(chatActions.endpointId, context.endpoint.id), eq(chatActions.kind, "slack_session_stop"))))
+        .toEqual([expect.objectContaining({ status: "processed", result: expect.objectContaining({ code: "slack_session_stop_cancelled" }) })]);
+      expect(await readAgentWakeupRequestsById(context.action.id)).toEqual([expect.objectContaining({ status: "cancelled" })]);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({
+        payload: { text: "This follow-up was not started. Open the task in Paperclip for details.", progressState: "failed" },
+      })]));
+      expect(errors).toEqual([]);
+    } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
   it.each(["manual", "scoped"] as const)("advances pre-run FIFO selection past a suppressed page with microsecond timestamps (%s)", async mode => {
     const context = await deferredChatQueueFixture();
     const rejectedIds = Array.from({ length: 50 }, () => randomUUID());

server/src/app.ts

@@ -196,6 +196,7 @@ import { toolAccessService } from "./services/tool-access.js";
 import { chatChannelService } from "./services/chat-channels.js";
 import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js";
 import { nativeChatProgressWork } from "./services/chat-native-progress-work.js";
+import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js";
 import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js";
 import {
   createCoalescedAsyncTrigger,
@@ -1288,6 +1289,7 @@ export async function createApp(
   chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+  chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatRunMilestoneWork(db, opts.authPublicBaseUrl).register(
     deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(),
   );

server/src/modules/wake-queue/adapters/postgres.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js";
 import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
 import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js";
@@ -312,6 +313,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
     },
 
     async cancelDeferredWake({ companyId, wakeId, reason, now }) {
+      await notifyChatInboundWakeNotices(tx, { companyId });
       const rows = await tx
         .update(agentWakeupRequests)
         .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now })
@@ -343,6 +345,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
     },
 
     async failDeferredWake({ companyId, wakeId, now }) {
+      await notifyChatInboundWakeNotices(tx, { companyId });
       const rows = await tx
         .update(agentWakeupRequests)
         .set({

server/src/modules/wake-queue/adapters/queued-comment-postgres.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js";
 import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js";
 import { and, eq, inArray } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -94,6 +95,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
     },
 
     async cancelWake({ wakeId, reason, now }) {
+      await notifyChatInboundWakeNotices(tx, { companyId, issueId });
       await tx
         .update(agentWakeupRequests)
         .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now })

server/src/routes/issues.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js";
 import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js";
 import { parseObject } from "../adapters/utils.js";
 import { prepareRestrictedWorkspaceRetry } from "../services/restricted-workspace-retry.js";
@@ -13373,14 +13374,18 @@ export function issueRoutes(
       const subjectId = req.body.subjectId as string;
       await assertIssueGrantSubjectExists(issue.companyId, subjectType, subjectId);
       const actor = getActorInfo(req);
-      const result = await db.transaction((tx) => ensureActiveIssueAccessGrant(tx, {
-        issueId: managedIssue.id,
-        subjectType,
-        subjectId,
-        source: "explicit",
-        grantedByUserId: actor.actorType === "user" ? actor.actorId : null,
-        grantedByAgentId: actor.agentId,
-      }));
+      const result = await db.transaction(async (tx) => {
+        // Grants on an ancestor can restore current chat authority on descendants.
+        await notifyChatInboundWakeNotices(tx, { companyId: issue.companyId });
+        return ensureActiveIssueAccessGrant(tx, {
+          issueId: managedIssue.id,
+          subjectType,
+          subjectId,
+          source: "explicit",
+          grantedByUserId: actor.actorType === "user" ? actor.actorId : null,
+          grantedByAgentId: actor.agentId,
+        });
+      });
       if (result.created) {
         await logActivity(db, {
           companyId: issue.companyId,

server/src/services/chat-channels.ts

@@ -3352,7 +3352,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
 
   async function processPublicationMaintenance(limit = 25) {
     scheduleGitHubMaintenance();
-    await enqueueInboundWakeupPublications();
+    // Normal inbound notices are commit-driven. Failed-run retries retain their
+    // separate maintenance owner until its own producer audit is complete.
+    await enqueueFailedChatRetryPublications();
     await reconcileTerminalConfirmationActions(limit);
     await processPendingInteractionWakeups(limit);
   }

server/src/services/chat-channels/inbound/wakeups.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js";
 import { DELIVERY_PROCESSING_STALE_MS } from "./delivery-ledger.js";
 import type { EndpointRow, DbOrTransaction } from "../runtime-types.js";
 import { notifyChatRunMilestones } from "../../chat-run-milestone-notifications.js";
@@ -113,6 +114,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) {
         await notifyChatDeliveryWork(tx);
         if (action.status === "processed") {
           await notifyChatRunMilestones(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) });
+        } else if (action.status === "failed") {
+          await notifyChatInboundWakeNotices(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) });
         }
         const acceptedAt = new Date();
         const accepted = await tx
@@ -255,6 +258,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) {
         await notifyChatDeliveryWork(tx);
         if (status === "processed") {
           await notifyChatRunMilestones(tx, { companyId: claimed.companyId, issueId: request!.issueId });
+        } else if (status === "failed" && request) {
+          await notifyChatInboundWakeNotices(tx, { companyId: claimed.companyId, issueId: request.issueId });
         }
         const changed = await tx
           .update(chatActions)

server/src/services/chat-channels/slack/session-stop.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js";
 import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js";
 
 import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js";
@@ -584,6 +585,7 @@ export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions)
           ["queued", "deferred_issue_execution"].includes(wakeup.status) &&
           !issue.executionRunId
         ) {
+          await notifyChatInboundWakeNotices(tx, { companyId: current.companyId, issueId: issue.id });
           await tx
             .update(agentWakeupRequests)
             .set({

server/src/services/companies.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { isRunnerApiResponseAsset } from "./asset-content-access.js";
@@ -65,6 +66,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
       ));
 
+    await notifyChatInboundWakeNotices(tx, { companyId: id });
     await tx
       .update(agentWakeupRequests)
       .set({
@@ -332,6 +334,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyCompanyRetentionWork(tx, id);
+          await notifyChatInboundWakeNotices(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
         }
         const updated = await tx
@@ -484,6 +487,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         if (!wasAlreadyArchived) {
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyCompanyRetentionWork(tx, id);
+          await notifyChatInboundWakeNotices(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
           await tx
             .update(companies)

server/src/services/heartbeat/run-control.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
 import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
 import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js";
 import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, readNativeLocalProcessStop, recordedLocalExecutionStopped } from "../native-local-process-stop.js";
@@ -976,6 +977,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
 
     if (wakeupIds.length === 0) return 0;
 
+    await notifyChatInboundWakeNotices(tx, { companyId: scope.companyId });
     await tx
       .update(agentWakeupRequests)
       .set({
@@ -1516,35 +1518,29 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
     );
     if (uniqueAgentIds.length === 0) return 0;
 
-    const now = new Date();
-    const wakeupIds = await db
-      .select({ id: agentWakeupRequests.id })
-      .from(agentWakeupRequests)
-      .where(
-        and(
+    return db.transaction(async tx => {
+      const wakeups = await tx
+        .select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId })
+        .from(agentWakeupRequests)
+        .where(and(
           inArray(agentWakeupRequests.agentId, uniqueAgentIds),
-          inArray(agentWakeupRequests.status, [
-            "queued",
-            "deferred_issue_execution",
-          ]),
-          sql`${agentWakeupRequests.runId} is null`,
-        ),
-      )
-      .then((rows) => rows.map((row) => row.id));
-
-    if (wakeupIds.length === 0) return 0;
-
-    await db
-      .update(agentWakeupRequests)
-      .set({
-        status: "cancelled",
-        finishedAt: now,
-        error: reason,
-        updatedAt: now,
-      })
-      .where(inArray(agentWakeupRequests.id, wakeupIds));
-
-    return wakeupIds.length;
+          inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
+          isNull(agentWakeupRequests.runId),
+        ));
+      if (!wakeups.length) return 0;
+      for (const companyId of new Set(wakeups.map(row => row.companyId))) {
+        await notifyChatInboundWakeNotices(tx, { companyId });
+      }
+      const now = new Date();
+      const cancelled = await tx.update(agentWakeupRequests).set({
+        status: "cancelled", finishedAt: now, error: reason, updatedAt: now,
+      }).where(and(
+        inArray(agentWakeupRequests.id, wakeups.map(row => row.id)),
+        inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
+        isNull(agentWakeupRequests.runId),
+      )).returning({ id: agentWakeupRequests.id });
+      return cancelled.length;
+    });
   }
 
   async function cancelInvocationsForAgentsInternal(

server/src/services/issues.ts

@@ -11194,6 +11194,9 @@ export function issueService(db: Db) {
           });
         }
         if (changesPrivacy) {
+          // Privacy inherited through task ancestry can authorize saved notices
+          // on descendants as well as this task.
+          await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId });
           const nextProjectId = issueData.projectId !== undefined ? issueData.projectId : receiptExisting.projectId;
           const [privacyProject] = nextProjectId ? await tx.select().from(projects)
             .where(and(eq(projects.id, nextProjectId), eq(projects.companyId, existing.companyId))) : [];
@@ -11275,6 +11278,7 @@ export function issueService(db: Db) {
           // Invalidate human direction at the common assignment boundary, including
           // plugin/service writes that do not go through HTTP run cancellation.
           // Keep the requester attribution for audit; cancellation revokes its use.
+          await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId });
           await tx.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() })
             .where(and(eq(agentWakeupRequests.companyId, receiptExisting.companyId),
               eq(agentWakeupRequests.requestedByActorType, "user"),
@@ -11303,6 +11307,9 @@ export function issueService(db: Db) {
         if (patch.status === "done" || patch.status === "cancelled") {
           await notifyDecisionTargetWork(tx, `issue:${id}`);
         }
+        if (patch.status !== undefined && patch.status !== receiptExisting.status) {
+          await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id });
+        }
         if (patch.hiddenAt === null && receiptExisting.hiddenAt !== null) {
           await notifyChatRunMilestones(tx, { companyId: receiptExisting.companyId, issueId: id });
         }

server/src/services/projects.ts

@@ -1,3 +1,4 @@
+import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { listProjectWorkspaceProviders, resolveProjectWorkspaceProvider } from "./project-workspace-provider.js";
 import { retireRemovedPluginWorkspaceDependents } from "./plugin-workspaces.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -80,12 +81,11 @@ export async function ensureProjectAccessMember(
     subjectId: string;
   },
 ) {
-  return dbOrTx
-    .insert(projectAccessMembers)
-    .values(input)
-    .onConflictDoNothing()
-    .returning()
-    .then((rows: Array<typeof projectAccessMembers.$inferSelect>) => rows[0] ?? null);
+  return dbOrTx.transaction(async (tx: Db) => {
+    await notifyChatInboundWakeNotices(tx, { companyId: input.companyId });
+    return tx.insert(projectAccessMembers).values(input).onConflictDoNothing().returning()
+      .then(rows => rows[0] ?? null);
+  });
 }
 
 /**
@@ -1013,6 +1013,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool
 
       const row = await db.transaction(async (tx) => {
         if (updates.visibility !== undefined) {
+          await notifyChatInboundWakeNotices(tx, { companyId: existingProject.companyId });
           await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`issue-privacy-tree:${existingProject.companyId}`}, 0))`);
         }
         const [updated] = await tx.update(projects).set(updates).where(eq(projects.id, id)).returning();