mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): wake chat milestones when publication authority returns
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
0e5591054a
commit
0f38d1079d
16 files changed
+227
-40
No files matched your search
@@ -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
|
classification callers reuse their issue identity, with no new queue table or
|
||||||
provider policy in shared code. Authority/configuration audit and production
|
provider policy in shared code. Authority/configuration audit and production
|
||||||
wiring remain next; the original full stack and remote gates remain incomplete.
|
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.
|
||||||
@@ -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
|
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
|
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.
|
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.
|
||||||
@@ -47,7 +47,7 @@ describe("chat action scheduling", () => {
|
|||||||
const work = s.start();
|
const work = s.start();
|
||||||
await vi.advanceTimersByTimeAsync(1);
|
await vi.advanceTimersByTimeAsync(1);
|
||||||
expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(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");
|
expect(s.service.scheduleTelegramSubscriptionDiscovery).toHaveBeenCalledWith("connection-fixture");
|
||||||
s.service.processTelegramSubscriptionDiscovery.mockRejectedValueOnce(new Error("discovery unavailable"));
|
s.service.processTelegramSubscriptionDiscovery.mockRejectedValueOnce(new Error("discovery unavailable"));
|
||||||
release();
|
release();
|
||||||
@@ -87,7 +87,7 @@ describe("chat action scheduling", () => {
|
|||||||
for (const run of s.runs) expect(run).toHaveBeenCalledTimes(2);
|
for (const run of s.runs) expect(run).toHaveBeenCalledTimes(2);
|
||||||
// Queue claims/settlements do not rediscover every configured endpoint.
|
// Queue claims/settlements do not rediscover every configured endpoint.
|
||||||
expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(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"));
|
||||||
await vi.advanceTimersByTimeAsync(1);
|
await vi.advanceTimersByTimeAsync(1);
|
||||||
for (const run of s.runs) expect(run).toHaveBeenCalledTimes(3);
|
for (const run of s.runs) expect(run).toHaveBeenCalledTimes(3);
|
||||||
expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(2);
|
expect(s.service.processTelegramSubscriptionDiscovery).toHaveBeenCalledTimes(2);
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import { accessService } from "../services/access.js";
|
||||||
import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||||
import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js";
|
import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js";
|
||||||
import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.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(); }
|
} 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<unknown>;
|
||||||
|
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 () => {
|
it("publishes working and the exact fresh final after pinned close without resurrecting the old source", async () => {
|
||||||
const f = await reopenedFixture();
|
const f = await reopenedFixture();
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
|
||||||
import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
|
import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
|
||||||
import { notifyChatEndpointWork } from "./chat-work-notifications.js";
|
import { notifyChatEndpointWork } from "./chat-work-notifications.js";
|
||||||
import { notifyToolConnectionHealth } from "./tool-connection-health-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.length > 0
|
||||||
&& grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)),
|
&& 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 (connection.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: connection.id });
|
||||||
if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id });
|
if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id });
|
||||||
await tx.update(toolConnections).set({
|
await tx.update(toolConnections).set({
|
||||||
@@ -554,6 +555,11 @@ export function accessService(db: Db) {
|
|||||||
data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole;
|
data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole;
|
||||||
const nextStatus = data.status ?? existing.status;
|
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 (
|
if (
|
||||||
existing.principalType === "user" &&
|
existing.principalType === "user" &&
|
||||||
existing.status === "active" &&
|
existing.status === "active" &&
|
||||||
@@ -886,6 +892,7 @@ export function accessService(db: Db) {
|
|||||||
const existingMembership = existingByCompany.get(companyId);
|
const existingMembership = existingByCompany.get(companyId);
|
||||||
if (existingMembership) {
|
if (existingMembership) {
|
||||||
if (existingMembership.status !== "active") {
|
if (existingMembership.status !== "active") {
|
||||||
|
if (existingMembership.membershipRole !== "viewer") await notifyChatRunMilestones(tx, { companyId });
|
||||||
await tx
|
await tx
|
||||||
.update(companyMemberships)
|
.update(companyMemberships)
|
||||||
.set({
|
.set({
|
||||||
@@ -897,6 +904,7 @@ export function accessService(db: Db) {
|
|||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
await notifyChatRunMilestones(tx, { companyId });
|
||||||
await tx.insert(companyMemberships).values({
|
await tx.insert(companyMemberships).values({
|
||||||
companyId,
|
companyId,
|
||||||
principalType: "user",
|
principalType: "user",
|
||||||
@@ -917,31 +925,43 @@ export function accessService(db: Db) {
|
|||||||
membershipRole: string | null = "member",
|
membershipRole: string | null = "member",
|
||||||
status: "pending" | "active" | "suspended" = "active",
|
status: "pending" | "active" | "suspended" = "active",
|
||||||
) {
|
) {
|
||||||
const existing = await getMembership(companyId, principalType, principalId);
|
return db.transaction(async tx => {
|
||||||
if (existing) {
|
const existing = await tx.select().from(companyMemberships).where(and(
|
||||||
if (existing.status !== status || existing.membershipRole !== membershipRole) {
|
eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalType, principalType),
|
||||||
const updated = await db
|
eq(companyMemberships.principalId, principalId),
|
||||||
.update(companyMemberships)
|
)).then(rows => rows[0] ?? null);
|
||||||
.set({ status, membershipRole, updatedAt: new Date() })
|
if (existing) {
|
||||||
.where(eq(companyMemberships.id, existing.id))
|
if (existing.status !== status || existing.membershipRole !== membershipRole) {
|
||||||
.returning()
|
if (principalType === "user" && status === "active" && membershipRole !== "viewer"
|
||||||
.then((rows) => rows[0] ?? null);
|
&& (existing.status !== "active" || existing.membershipRole === "viewer")) {
|
||||||
return updated ?? existing;
|
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
|
if (principalType === "user" && status === "active" && membershipRole !== "viewer") {
|
||||||
.insert(companyMemberships)
|
await notifyChatRunMilestones(tx, { companyId });
|
||||||
.values({
|
}
|
||||||
companyId,
|
return tx
|
||||||
principalType,
|
.insert(companyMemberships)
|
||||||
principalId,
|
.values({
|
||||||
status,
|
companyId,
|
||||||
membershipRole,
|
principalType,
|
||||||
})
|
principalId,
|
||||||
.returning()
|
status,
|
||||||
.then((rows) => rows[0]);
|
membershipRole,
|
||||||
|
})
|
||||||
|
.returning()
|
||||||
|
.then((rows) => rows[0]);
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
async function setPrincipalGrants(
|
async function setPrincipalGrants(
|
||||||
@@ -1122,6 +1142,11 @@ export function accessService(db: Db) {
|
|||||||
data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole;
|
data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole;
|
||||||
const nextStatus = data.status ?? existing.status;
|
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 (
|
if (
|
||||||
existing.principalType === "user" &&
|
existing.principalType === "user" &&
|
||||||
existing.status === "active" &&
|
existing.status === "active" &&
|
||||||
|
|||||||
@@ -182,7 +182,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
code: "chat_endpoint_not_active",
|
code: "chat_endpoint_not_active",
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
await tx
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -319,7 +319,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
inArray(chatDeliveries.state, ["received", "retry"]),
|
inArray(chatDeliveries.state, ["received", "retry"]),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
[activatedEndpoint] = await tx
|
[activatedEndpoint] = await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -367,7 +367,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
const failedAt = new Date();
|
const failedAt = new Date();
|
||||||
await db.transaction(async (tx) => {
|
await db.transaction(async (tx) => {
|
||||||
await credentialLease.assertOwned(tx);
|
await credentialLease.assertOwned(tx);
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
await tx
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -471,7 +471,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
code: "chat_endpoint_already_removed",
|
code: "chat_endpoint_already_removed",
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
await tx
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -703,7 +703,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
try {
|
try {
|
||||||
const claimed = await db.transaction(async (tx) => {
|
const claimed = await db.transaction(async (tx) => {
|
||||||
await credentialLease.assertOwned(tx);
|
await credentialLease.assertOwned(tx);
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
const rows = await tx
|
const rows = await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -783,7 +783,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
const preserveSlackVerification = isProvisionedSlackSetup(current.setup.slackSetupMethod)
|
const preserveSlackVerification = isProvisionedSlackSetup(current.setup.slackSetupMethod)
|
||||||
&& (current.setup as InternalSetupState).slackVerificationSigningFingerprint === createHash("sha256").update(credentials.signingSecret ?? "").digest("hex")
|
&& (current.setup as InternalSetupState).slackVerificationSigningFingerprint === createHash("sha256").update(credentials.signingSecret ?? "").digest("hex")
|
||||||
&& Boolean(observedSlackUrl && slackCallbackMatchesPublicUrl(observedSlackUrl, `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/slack`));
|
&& 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
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
@@ -869,7 +869,7 @@ export function createChatConfiguration(db: Db, options: ChatConfigurationOption
|
|||||||
await invalidateRuntime(endpoint.id).catch(() => undefined);
|
await invalidateRuntime(endpoint.id).catch(() => undefined);
|
||||||
await db.transaction(async (tx) => {
|
await db.transaction(async (tx) => {
|
||||||
await credentialLease.assertOwned(tx);
|
await credentialLease.assertOwned(tx);
|
||||||
await notifyChatEndpointWork(tx, endpoint.connectionId);
|
await notifyChatEndpointWork(tx, endpoint.connectionId, endpoint.companyId);
|
||||||
await tx
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -165,7 +165,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents
|
|||||||
current.setup.step === "complete" ? "active" : "verifying";
|
current.setup.step === "complete" ? "active" : "verifying";
|
||||||
await notifyChatReceiver(tx, current);
|
await notifyChatReceiver(tx, current);
|
||||||
await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id });
|
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
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
|
||||||
import { chatConnectionTransport } from "./provider-catalog.js";
|
import { chatConnectionTransport } from "./provider-catalog.js";
|
||||||
import { slackSetupState } from "./slack/setup.js";
|
import { slackSetupState } from "./slack/setup.js";
|
||||||
import { assertChatExecutionDefaultsAccess } from "../chat-execution-defaults.js";
|
import { assertChatExecutionDefaultsAccess } from "../chat-execution-defaults.js";
|
||||||
@@ -445,6 +446,11 @@ function providerSetupState(
|
|||||||
values.allowUnlinkedPeople = input.allowUnlinkedPeople;
|
values.allowUnlinkedPeople = input.allowUnlinkedPeople;
|
||||||
await db.transaction(async (tx) => {
|
await db.transaction(async (tx) => {
|
||||||
await credentialLease.assertOwned(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) {
|
if (input.executionDefaults !== undefined) {
|
||||||
await assertChatExecutionDefaultsAccess(db, {
|
await assertChatExecutionDefaultsAccess(db, {
|
||||||
companyId: existing.endpoint.companyId,
|
companyId: existing.endpoint.companyId,
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
|
||||||
import { githubIdentityFields } from "./github/identity-links.js";
|
import { githubIdentityFields } from "./github/identity-links.js";
|
||||||
import { createSlackIdentityLinks, type SlackIdentityEffectStager } from "./slack/identity-links.js";
|
import { createSlackIdentityLinks, type SlackIdentityEffectStager } from "./slack/identity-links.js";
|
||||||
import { createHash, randomBytes } from "node:crypto";
|
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
|
const confirmedLink = await tx
|
||||||
.update(chatIdentityLinks)
|
.update(chatIdentityLinks)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -32,7 +32,7 @@ export function createPhotonRuntimeCallbacks(db: Db, options: PhotonRuntimeOptio
|
|||||||
if (!current) return;
|
if (!current) return;
|
||||||
const connection = await tx.select().from(toolConnections).where(eq(toolConnections.id, current.connectionId)).then((rows) => rows[0]);
|
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");
|
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));
|
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));
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -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 } : {}) },
|
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) } : {}) },
|
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) {
|
if (bulk) {
|
||||||
// One atomic database update, independent of pagination/search and
|
// One atomic database update, independent of pagination/search and
|
||||||
// serialized with refresh, reconnect, and individual row changes.
|
// serialized with refresh, reconnect, and individual row changes.
|
||||||
@@ -500,7 +503,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc
|
|||||||
? "verifying"
|
? "verifying"
|
||||||
: currentEndpoint.status;
|
: currentEndpoint.status;
|
||||||
await notifyChatEndpointConfiguration(tx, currentEndpoint);
|
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
|
await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -297,7 +297,7 @@ export function createSetupProbes(db: Db, options: SetupProbesOptions) {
|
|||||||
)).limit(1).then((rows) => rows[0]);
|
)).limit(1).then((rows) => rows[0]);
|
||||||
if (!linked || !(await lockCurrentPrincipalAuthorization(tx, endpoint, linked.principalId)).allowed) throw forbidden("Your connected account is no longer authorized");
|
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
|
const [activated] = await tx
|
||||||
.update(chatEndpoints)
|
.update(chatEndpoints)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -42,7 +42,10 @@ export async function notifyChatVerificationWork(tx: object): Promise<void> {
|
|||||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatVerification);
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatVerification);
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function notifyChatEndpointWork(tx: object, connectionId: string): Promise<void> {
|
export async function notifyChatEndpointWork(tx: object, connectionId: string, companyId: string): Promise<void> {
|
||||||
|
// 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 notifyChatDeliveryWork(tx);
|
||||||
await notifyChatPublicationWork(tx);
|
await notifyChatPublicationWork(tx);
|
||||||
for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue);
|
for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue);
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
|
||||||
import { notifyApprovalWork } from "./approval-work-notifications.js";
|
import { notifyApprovalWork } from "./approval-work-notifications.js";
|
||||||
import { notifyDecisionTargetWork } from "./decision-target-work.js";
|
import { notifyDecisionTargetWork } from "./decision-target-work.js";
|
||||||
import { insertIssueComment } from "./issue-comment-writes.js";
|
import { insertIssueComment } from "./issue-comment-writes.js";
|
||||||
@@ -11301,6 +11302,9 @@ export function issueService(db: Db) {
|
|||||||
if (patch.status === "done" || patch.status === "cancelled") {
|
if (patch.status === "done" || patch.status === "cancelled") {
|
||||||
await notifyDecisionTargetWork(tx, `issue:${id}`);
|
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) {
|
if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) {
|
||||||
await notifyApprovalWork(tx, receiptExisting.companyId);
|
await notifyApprovalWork(tx, receiptExisting.companyId);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4642,7 +4642,7 @@ export function toolAccessService(
|
|||||||
.returning();
|
.returning();
|
||||||
const [row] = existing.transport === "chat_sdk"
|
const [row] = existing.transport === "chat_sdk"
|
||||||
? await db.transaction(async tx => {
|
? 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);
|
const rows = await write(tx as unknown as Db);
|
||||||
// Credential resolution must never observe the binding rebuild's
|
// Credential resolution must never observe the binding rebuild's
|
||||||
// delete/insert gap after a configuration wake.
|
// delete/insert gap after a configuration wake.
|
||||||
|
|||||||
@@ -539,7 +539,7 @@ async function removeConnection(
|
|||||||
.returning({ id: connectionTokenIssuances.id });
|
.returning({ id: connectionTokenIssuances.id });
|
||||||
|
|
||||||
const archived = await db.transaction(async (tx) => {
|
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 });
|
if (connection.transport === "voice") await notifyVoiceChange(tx, { kind: "connection", id: connection.id });
|
||||||
await notifyToolConnectionHealth(tx, connection);
|
await notifyToolConnectionHealth(tx, connection);
|
||||||
await tx.select({ id: toolApplications.id }).from(toolApplications)
|
await tx.select({ id: toolApplications.id }).from(toolApplications)
|
||||||
|
|||||||
Reference in new issue
Block a user