diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 0485fbf9638..bd37f78a858 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1677,3 +1677,14 @@ milestone cases, 89 recovery-action cases and server types/build pass. Four classification callers reuse their issue identity, with no new queue table or provider policy in shared code. Authority/configuration audit and production wiring remain next; the original full stack and remote gates remain incomplete. + + +Slice 28 now covers nine real authority-restoration owners: connection/resource +reach, unlinked policy, identity confirmation, four membership paths and task +visibility. Seven initial causal failures plus connection/resource reproductions +now pass; the invalid Discord group-reach fixture is explicitly retained in the +failure record. Final local selection: 50 chat integrations, 18 access tests, +six action-worker tests, server typecheck/build. No provider authorization was +relaxed. Cloud trusted-header membership sync and board claim remain alternate +writers to cover before production registration/poll removal. Older parent +findings and the complete original stack remain open. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index 05b654ac090..52ad57b1d50 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -1350,3 +1350,66 @@ register the lifecycle worker, retain the separate native-progress duty, remove only the lifecycle poll, and complete current-head CI/review. Fresh master 34409a92c4 is fetched but not yet reconciled into the stack. Older parent CI and voice security/deletion findings plus every original remaining reason stay open. + + +## Restoring current publication authority + +Post-control lineage reads current connection enablement/status, resource reach, +endpoint reach/unlinked policy, linked human membership, and task visibility. +Restoring one of those facts can make a retained notice eligible without a run +transition. Nine real service-owner cases now cover that change after the worker +has exhausted startup: connection enablement, resource selection, unlinked reach, +identity confirmation, updateMember, updateMemberAndPermissions, ensureMembership, +setUserCompanyAccess, and task visibility. All keep the original authorization +reader and exact admitted source proof. No new run is dispatched to get a notice. + +Owner changes: +- The existing notifyChatEndpointWork contract now requires the already-known + company identity and also wakes milestones. Its configuration/status callers + pass that identity; ordinary health-only callbacks still use the narrower + notifyChatEndpointConfiguration/notifyChatConnectionConfiguration contracts. + This is a bounded wake on an actual configuration operation, not a health poll. +- Endpoint policy edits signal only newly enabled direct/group/unlinked reach. + Identity confirmation signals inside its existing advisory/row-lock transaction. +- Resource selection signals only a net disabled-to-enabled grant, covering both + individual and bulk GitHub selection through their common owner. Unchanged + inventory/defaults/selection does not create milestone work. +- Membership mutation owners signal human activation or viewer-to-active-role + restoration before writing. ensureMembership now owns its read/write transaction; + agent memberships and unchanged human state do not emit milestone intents. +- The issue update owner signals only hidden-to-visible restoration. Ordinary + chat-bound assignee changes are rejected by the existing chat_binding_agent_locked + guard, not treated as a supported restoration path. + +Evidence and corrected failures: +- /tmp/sleep28-authority-before.log: seven actual missed-wake failures, plus an + invalid Discord group-reach test. Discord channels use resource.enabled rather + than allowGroupChats; that fixture was already eligible before restoration. + Do not count that invalid failure as a production defect. +- /tmp/sleep28-authority-tests.log: the seven valid cases pass after repair; the + incorrect reach test still fails its precondition. It was replaced with actual + connection enablement, which fails before repair in + /tmp/sleep28-connection-authority-before.log. +- /tmp/sleep28-resource-selection-before.log: real resource grant restoration + misses its notice. Existing NOWAIT retry logs are retained; the assertion + failure is absence of the terminally eligible publication. +- /tmp/sleep28-authority-final-acceptance.log: all 50 selected integration cases + pass (the full post-close authority group plus ingress/revocation/Telegram + migration). Nine restoration cases include unchanged replay quietness where + supported; one-time link confirmation retains its existing consumed-token + behavior. Explicit connection updates retain their configuration wake. +- /tmp/sleep28-access-acceptance.log: all 18 access service regressions pass. +- /tmp/sleep28-authority-action-tests.log: all six action-worker cases pass. +- Initial typecheck caught a caller using connection.companyId that its query + did not select. It now uses the owning operation's companyId. Final server + typecheck/build pass in /tmp/sleep28-authority-final-{typecheck,build}.log. + +Alternate human-membership producers remain: middleware/auth.ts Cloud trusted- +header synchronization and board-claim.ts. Startup local-board insertion runs +before worker reconstruction; compatibility backfill only inserts agent members. +Cloud sign-in and board claim can restore humans while a worker is running and +must be covered before removing its poll. Credential rotation advances the +runtime fence; old sources must not regain authority from a new credential +fingerprint. Keep the existing generation/chronology denial tests. Remaining +producer dispositions, production wiring, full remote gates and original stack +scope are still required; this checkpoint is not a completed sleep migration. diff --git a/server/src/__tests__/chat-action-work.test.ts b/server/src/__tests__/chat-action-work.test.ts index 067d6d54e1b..7576c93d4aa 100644 --- a/server/src/__tests__/chat-action-work.test.ts +++ b/server/src/__tests__/chat-action-work.test.ts @@ -47,7 +47,7 @@ describe("chat action scheduling", () => { const work = s.start(); await vi.advanceTimersByTimeAsync(1); expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(1); - await s.owner.transaction(tx => notifyChatEndpointWork(tx, "connection-fixture")); + await s.owner.transaction(tx => notifyChatEndpointWork(tx, "connection-fixture", "company-fixture")); expect(s.service.scheduleTelegramSubscriptionDiscovery).toHaveBeenCalledWith("connection-fixture"); s.service.processTelegramSubscriptionDiscovery.mockRejectedValueOnce(new Error("discovery unavailable")); release(); @@ -87,7 +87,7 @@ describe("chat action scheduling", () => { for (const run of s.runs) expect(run).toHaveBeenCalledTimes(2); // Queue claims/settlements do not rediscover every configured endpoint. expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(1); - await s.owner.transaction(tx => notifyChatEndpointWork(tx, "connection-fixture")); + await s.owner.transaction(tx => notifyChatEndpointWork(tx, "connection-fixture", "company-fixture")); await vi.advanceTimersByTimeAsync(1); for (const run of s.runs) expect(run).toHaveBeenCalledTimes(3); expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(2); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index f57bca86b12..45d6c3a3640 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { accessService } from "../services/access.js"; import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js"; import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; @@ -52620,6 +52621,75 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await coordinator.stop(); await f.close(); } }); + it.each(["connection", "resource", "unlinked", "identity", "member", "member_permissions", "ensure_membership", "company_access", "visibility"] as const)( + "wakes a post-close milestone when %s authority is restored", async owner => { + const f = await reopenedFixture(); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const notice = () => db.select().from(chatPublications).where(and( + eq(chatPublications.issueId, f.conversation.issueId), + eq(chatPublications.idempotencyKey, `run:${f.fresh.runId}:working:${f.endpoint.id}`), + )); + const access = accessService(db); + const membership = (await db.select().from(companyMemberships).where(and( + eq(companyMemberships.companyId, f.fixture.companyId), eq(companyMemberships.principalId, "owner-user"), + )))[0]; + try { + let restore: () => Promise; + if (owner === "connection") { + await toolAccessService(db).updateConnection(f.endpoint.connectionId, { enabled: false }); + restore = () => toolAccessService(db).updateConnection(f.endpoint.connectionId, { enabled: true }); + } else if (owner === "resource") { + const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.id, f.conversation.id)); + expect(conversation.resourceId).not.toBeNull(); + await f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: false }], "owner-user"); + restore = () => f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: true }], "owner-user"); + } else if (owner === "identity" || owner === "unlinked") { + await f.service.update(f.endpoint.id, { allowUnlinkedPeople: false }, "owner-user"); + await f.service.revokeLink(f.endpoint.id, f.fresh.action.principalId!); + if (owner === "unlinked") { + restore = () => f.service.update(f.endpoint.id, { allowUnlinkedPeople: true }, "owner-user"); + } else { + const intent = await f.service.createLinkIntent(f.endpoint.id, f.fresh.action.principalId!, 1800); + restore = () => f.service.confirmIdentityLink(new URL(intent.confirmationUrl).searchParams.get("token")!, "owner-user"); + } + } else if (owner === "visibility") { + await issueService(db).update(f.conversation.issueId, { hiddenAt: new Date() }); + restore = () => issueService(db).update(f.conversation.issueId, { hiddenAt: null }); + } else { + // Model existing denied authority without exercising revocation's + // unrelated credential cleanup; restoration uses its actual owner. + await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.id, membership.id)); + if (owner === "member") restore = () => access.updateMember(f.fixture.companyId, membership.id, { status: "active" }); + else if (owner === "member_permissions") restore = () => access.updateMemberAndPermissions(f.fixture.companyId, membership.id, { status: "active", grants: [] }, "owner-user"); + else if (owner === "ensure_membership") restore = () => access.ensureMembership(f.fixture.companyId, "user", "owner-user", membership.membershipRole, "active"); + else { + const existing = await db.select().from(companyMemberships).where(and( + eq(companyMemberships.principalType, "user"), eq(companyMemberships.principalId, "owner-user"), + )); + restore = () => access.setUserCompanyAccess("owner-user", existing.map(row => row.companyId)); + } + } + await chatRunMilestoneWork(db).register(coordinator, () => true).ready; + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull()); + expect(await notice()).toEqual([]); + await restore(); + await expect.poll(notice, { timeout: 5000 }).toEqual([ + expect.objectContaining({ idempotencyKey: `run:${f.fresh.runId}:working:${f.endpoint.id}` }), + ]); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull()); + if (owner !== "identity" && owner !== "connection") { + const notices: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatRunMilestones, event => notices.push(event)); + try { await restore(); expect(notices).toEqual([]); } + finally { unsubscribe(); } + expect(await notice()).toHaveLength(1); + } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); await f.close(); } + }, + ); + it("publishes working and the exact fresh final after pinned close without resurrecting the old source", async () => { const f = await reopenedFixture(); try { diff --git a/server/src/services/access.ts b/server/src/services/access.ts index 0a2131d3369..2f7384483da 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js"; import { notifyChatEndpointWork } from "./chat-work-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; @@ -328,7 +329,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.length > 0 && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); - if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id, companyId); if (connection.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: connection.id }); if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id }); await tx.update(toolConnections).set({ @@ -554,6 +555,11 @@ export function accessService(db: Db) { data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole; const nextStatus = data.status ?? existing.status; + if (existing.principalType === "user" && nextStatus === "active" && nextMembershipRole !== "viewer" + && (existing.status !== "active" || existing.membershipRole === "viewer")) { + await notifyChatRunMilestones(tx, { companyId }); + } + if ( existing.principalType === "user" && existing.status === "active" && @@ -886,6 +892,7 @@ export function accessService(db: Db) { const existingMembership = existingByCompany.get(companyId); if (existingMembership) { if (existingMembership.status !== "active") { + if (existingMembership.membershipRole !== "viewer") await notifyChatRunMilestones(tx, { companyId }); await tx .update(companyMemberships) .set({ @@ -897,6 +904,7 @@ export function accessService(db: Db) { } continue; } + await notifyChatRunMilestones(tx, { companyId }); await tx.insert(companyMemberships).values({ companyId, principalType: "user", @@ -917,31 +925,43 @@ export function accessService(db: Db) { membershipRole: string | null = "member", status: "pending" | "active" | "suspended" = "active", ) { - const existing = await getMembership(companyId, principalType, principalId); - if (existing) { - if (existing.status !== status || existing.membershipRole !== membershipRole) { - const updated = await db - .update(companyMemberships) - .set({ status, membershipRole, updatedAt: new Date() }) - .where(eq(companyMemberships.id, existing.id)) - .returning() - .then((rows) => rows[0] ?? null); - return updated ?? existing; + return db.transaction(async tx => { + const existing = await tx.select().from(companyMemberships).where(and( + eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalType, principalType), + eq(companyMemberships.principalId, principalId), + )).then(rows => rows[0] ?? null); + if (existing) { + if (existing.status !== status || existing.membershipRole !== membershipRole) { + if (principalType === "user" && status === "active" && membershipRole !== "viewer" + && (existing.status !== "active" || existing.membershipRole === "viewer")) { + await notifyChatRunMilestones(tx, { companyId }); + } + const updated = await tx + .update(companyMemberships) + .set({ status, membershipRole, updatedAt: new Date() }) + .where(eq(companyMemberships.id, existing.id)) + .returning() + .then((rows) => rows[0] ?? null); + return updated ?? existing; + } + return existing; } - return existing; - } - return db - .insert(companyMemberships) - .values({ - companyId, - principalType, - principalId, - status, - membershipRole, - }) - .returning() - .then((rows) => rows[0]); + if (principalType === "user" && status === "active" && membershipRole !== "viewer") { + await notifyChatRunMilestones(tx, { companyId }); + } + return tx + .insert(companyMemberships) + .values({ + companyId, + principalType, + principalId, + status, + membershipRole, + }) + .returning() + .then((rows) => rows[0]); + }); } async function setPrincipalGrants( @@ -1122,6 +1142,11 @@ export function accessService(db: Db) { data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole; const nextStatus = data.status ?? existing.status; + if (existing.principalType === "user" && nextStatus === "active" && nextMembershipRole !== "viewer" + && (existing.status !== "active" || existing.membershipRole === "viewer")) { + await notifyChatRunMilestones(tx, { companyId }); + } + if ( existing.principalType === "user" && existing.status === "active" && diff --git a/server/src/services/chat-channels/configuration.ts b/server/src/services/chat-channels/configuration.ts index 2d58158c7f7..9fcae40c323 100644 --- a/server/src/services/chat-channels/configuration.ts +++ b/server/src/services/chat-channels/configuration.ts @@ -182,7 +182,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption code: "chat_endpoint_not_active", }); } - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) .set({ @@ -319,7 +319,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption inArray(chatDeliveries.state, ["received", "retry"]), ), ); - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); [activatedEndpoint] = await tx .update(chatEndpoints) .set({ @@ -367,7 +367,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption const failedAt = new Date(); await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) .set({ @@ -471,7 +471,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption code: "chat_endpoint_already_removed", }); } - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) .set({ @@ -703,7 +703,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption try { const claimed = await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); const rows = await tx .update(chatEndpoints) .set({ @@ -783,7 +783,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption const preserveSlackVerification = isProvisionedSlackSetup(current.setup.slackSetupMethod) && (current.setup as InternalSetupState).slackVerificationSigningFingerprint === createHash("sha256").update(credentials.signingSecret ?? "").digest("hex") && Boolean(observedSlackUrl && slackCallbackMatchesPublicUrl(observedSlackUrl, `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/slack`)); - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) .set({ @@ -869,7 +869,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption await invalidateRuntime(endpoint.id).catch(() => undefined); await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/discord/gateway-events.ts b/server/src/services/chat-channels/discord/gateway-events.ts index 3ed17aa6aa3..f59e1c709ed 100644 --- a/server/src/services/chat-channels/discord/gateway-events.ts +++ b/server/src/services/chat-channels/discord/gateway-events.ts @@ -165,7 +165,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents current.setup.step === "complete" ? "active" : "verifying"; await notifyChatReceiver(tx, current); await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); - if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId); + if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId, current.companyId); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/endpoints.ts b/server/src/services/chat-channels/endpoints.ts index a59879cfb09..ffb46081511 100644 --- a/server/src/services/chat-channels/endpoints.ts +++ b/server/src/services/chat-channels/endpoints.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { chatConnectionTransport } from "./provider-catalog.js"; import { slackSetupState } from "./slack/setup.js"; import { assertChatExecutionDefaultsAccess } from "../chat-execution-defaults.js"; @@ -445,6 +446,11 @@ function providerSetupState( values.allowUnlinkedPeople = input.allowUnlinkedPeople; await db.transaction(async (tx) => { await credentialLease.assertOwned(tx); + if ((input.allowDirectMessages === true && !existing.endpoint.allowDirectMessages) + || (input.allowGroupChats === true && !existing.endpoint.allowGroupChats) + || (input.allowUnlinkedPeople === true && !existing.endpoint.allowUnlinkedPeople)) { + await notifyChatRunMilestones(tx, { companyId: existing.endpoint.companyId }); + } if (input.executionDefaults !== undefined) { await assertChatExecutionDefaultsAccess(db, { companyId: existing.endpoint.companyId, diff --git a/server/src/services/chat-channels/identity-links.ts b/server/src/services/chat-channels/identity-links.ts index b45ee22db66..abbd2340c94 100644 --- a/server/src/services/chat-channels/identity-links.ts +++ b/server/src/services/chat-channels/identity-links.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { githubIdentityFields } from "./github/identity-links.js"; import { createSlackIdentityLinks, type SlackIdentityEffectStager } from "./slack/identity-links.js"; import { createHash, randomBytes } from "node:crypto"; @@ -319,6 +320,7 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions }, ); } + await notifyChatRunMilestones(tx, { companyId: currentLink.companyId }); const confirmedLink = await tx .update(chatIdentityLinks) .set({ diff --git a/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts b/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts index 7b7e5557fae..414e070a844 100644 --- a/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts +++ b/server/src/services/chat-channels/imessage-photon/runtime-callbacks.ts @@ -32,7 +32,7 @@ export function createPhotonRuntimeCallbacks(db: Db, options: PhotonRuntimeOptio if (!current) return; const connection = await tx.select().from(toolConnections).where(eq(toolConnections.id, current.connectionId)).then((rows) => rows[0]); if (!connection?.enabled || connection.healthStatus === "failed") throw new Error("Photon endpoint requires operator recovery"); - await notifyChatEndpointWork(tx, connectionId); + await notifyChatEndpointWork(tx, connectionId, companyId); await tx.update(chatEndpoints).set({ status: current.setup.step === "complete" ? "active" : "verifying", healthMessage: "Photon receiver connected", lastError: null, updatedAt: new Date() }).where(eq(chatEndpoints.id, endpointId)); }); }, diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 448aa463e9e..17873b2f487 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -284,6 +284,9 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc before: { enabled: row.enabled, ...(JSON.stringify(row.executionDefaults) !== JSON.stringify(finalDefaults.get(row.id)) ? { executionDefaults: row.executionDefaults } : {}) }, after: { enabled: finalEnabled.get(row.id)!, ...(JSON.stringify(row.executionDefaults) !== JSON.stringify(finalDefaults.get(row.id)) ? { executionDefaults: finalDefaults.get(row.id) } : {}) }, })); + if (changes.some(change => !change.before.enabled && change.after.enabled)) { + await notifyChatRunMilestones(tx, { companyId: endpoint.companyId }); + } if (bulk) { // One atomic database update, independent of pagination/search and // serialized with refresh, reconnect, and individual row changes. @@ -500,7 +503,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc ? "verifying" : currentEndpoint.status; await notifyChatEndpointConfiguration(tx, currentEndpoint); - if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId); + if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId, currentEndpoint.companyId); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/setup-probes.ts b/server/src/services/chat-channels/setup-probes.ts index d5ae65d845c..2a86c486369 100644 --- a/server/src/services/chat-channels/setup-probes.ts +++ b/server/src/services/chat-channels/setup-probes.ts @@ -297,7 +297,7 @@ export function createSetupProbes(db: Db, options: SetupProbesOptions) { )).limit(1).then((rows) => rows[0]); if (!linked || !(await lockCurrentPrincipalAuthorization(tx, endpoint, linked.principalId)).allowed) throw forbidden("Your connected account is no longer authorized"); } - await notifyChatEndpointWork(tx, endpoint.connectionId); + await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId); const [activated] = await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 7a03bbcef51..5e0d05262c9 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -42,7 +42,10 @@ export async function notifyChatVerificationWork(tx: object): Promise { await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatVerification); } -export async function notifyChatEndpointWork(tx: object, connectionId: string): Promise { +export async function notifyChatEndpointWork(tx: object, connectionId: string, companyId: string): Promise { + // Configuration/status transitions can restore post-control publication + // authority. Ordinary health checks use the narrower configuration signals. + await notifyChatRunMilestones(tx, { companyId }); await notifyChatDeliveryWork(tx); await notifyChatPublicationWork(tx); for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 2964c3a122a..f864bdf0c57 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { insertIssueComment } from "./issue-comment-writes.js"; @@ -11301,6 +11302,9 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if (patch.hiddenAt === null && receiptExisting.hiddenAt !== null) { + await notifyChatRunMilestones(tx, { companyId: receiptExisting.companyId, issueId: id }); + } if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) { await notifyApprovalWork(tx, receiptExisting.companyId); } diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index ec0c1169df0..362e60bd00a 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -4642,7 +4642,7 @@ export function toolAccessService( .returning(); const [row] = existing.transport === "chat_sdk" ? await db.transaction(async tx => { - await notifyChatEndpointWork(tx, existing.id); + await notifyChatEndpointWork(tx, existing.id, existing.companyId); const rows = await write(tx as unknown as Db); // Credential resolution must never observe the binding rebuild's // delete/insert gap after a configuration wake. diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 2aa97e7ca50..c78459c78d7 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -539,7 +539,7 @@ async function removeConnection( .returning({ id: connectionTokenIssuances.id }); const archived = await db.transaction(async (tx) => { - if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id, connection.companyId); if (connection.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: connection.id }); await notifyToolConnectionHealth(tx, connection); await tx.select({ id: toolApplications.id }).from(toolApplications)