fix(server): wake chat milestones from inbound admission

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 10:38:24 -05:00
1 parent cc78a3c8b2
commit f582d65f66
4 files changed
+85

No files matched your search

@@ -1632,3 +1632,12 @@ Three local cancellation diagnostic runs passed, including full projects coverag
and a database-only delay, without proving the CI deadlock's competing owner.
Those failures, parent security/deletion findings, all other original slices,
fresh remote review and reconciliation with master 7ebf98eafd remain open.
Slice 28 ordinary ingress now wakes existing milestone debt before reactivating
conversation bindings and writing inbound links. A real Slack ingress regression
failed before repair and passes without depending on a new run being created.
All 12 selected ingress/revocation/Telegram migration cases and server types/build
pass. The milestone audit records evidence, isolated DB, and remaining authority/
lineage producers. This is another producer checkpoint in draft #17606, not
production polling removal or completion of the full requested stack.
@@ -1187,3 +1187,36 @@ E2E failures; no blind retry or cancellation-error masking was applied.
Normal inbound conversation/link activation and remaining authority/lineage
producer dispositions still gate production registration and polling removal.
Fresh master is 7ebf98eafd; final base reconciliation remains outstanding.
## Ordinary inbound conversation activation
The actual Slack ingress owner reproduces another missing wake: a prior failed
run has valid inbound lineage, but its unavailable conversation is restored by a
new message after the projector exhausted its startup work. The harness accepts
the new wake without creating a new run, so this does not rely on later queue
admission to hide the gap. Before repair the expected failed-run publication
never appears (/tmp/sleep28-inbound-before.log).
persistTaskMutation now registers the existing issue scope before its mutations,
or the newly created issue scope before inserting its conversation. The same
transaction creates the inbound link and reactivates the binding. Existing
endpoint/resource/principal authorization runs before this owner; provider calls
remain outside it. This is a direct use of the shared transaction intent helper,
not a new wrapper or background scan. No business policy or public API changes.
The regression passes (/tmp/sleep28-inbound-tests.log), and the combined selection
passes 12 cases: ordinary ingress reactivation, three existing Slack/Teams/Telegram
reach-revocation cases, and eight Telegram migration cases
(/tmp/sleep28-inbound-acceptance.log; 1,419 deselected). Server production
typecheck/build pass in /tmp/sleep28-inbound-{typecheck,build}.log. The isolated
DB is paperclip_sleep28_inbound_20261025_test. Existing resource and lifecycle
proofs remain recorded at their prior checkpoints; the whole integration file
was not run locally for this edit.
Remaining audit: alternate monitor/wakeup receipt rewrites, board reply identity,
provider-specific configuration/authority, and any remaining direct lineage
writers. Endpoint update does not assign a new agent or change publication mode;
its unrelated health/configuration writes must not cause blanket milestone scans.
Production registration/poll removal, complete remote gates, prior parent findings,
and the full original stack remain unfinished.
@@ -16,6 +16,7 @@ import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates } from
import type { PaperclipCloudConnector, SealedConnectorCredentials } from "../services/paperclip-cloud-connector.js";
import { buildSlackAppManifest, SLACK_MANAGER_SCOPES } from "@paperclipai/shared";
import { registerChatActionWork, dueChatAction } from "../services/chat-action-work.js";
import { chatRunMilestoneWork } from "../services/chat-run-milestone-work.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
import { notifyChatPublicationWork, notifyChatActionWork, CHAT_ACTION_QUEUES } from "../services/chat-work-notifications.js";
@@ -34442,6 +34443,43 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
]);
});
it("wakes an existing chat milestone when ordinary ingress reactivates its conversation", async () => {
const fixture = await seedCompany();
const { callbacks, endpoint, wakeup } = await configuredSlackEndpoint(fixture);
const thread = makeThread({ channelId: "C-REOPEN-MILESTONE", id: "slack:C-REOPEN-MILESTONE:4100.1", name: "reopen-milestone" });
const send = (id: string) => deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread,
message: makeMessage({ id, text: "@maya continue this conversation", mentioned: true }), trigger: "mention" });
await send("4100.1");
const [conversation] = await readChatConversationsByEndpointId(endpoint.id);
const [link] = await db.select().from(chatMessageLinks).where(and(
eq(chatMessageLinks.conversationId, conversation.id), eq(chatMessageLinks.providerMessageId, "4100.1")));
const runId = randomUUID();
await db.insert(heartbeatRuns).values({ id: runId, companyId: fixture.companyId, agentId: fixture.assignedAgentId,
status: "failed", errorCode: "low_trust_isolation_unavailable", updatedAt: new Date(),
contextSnapshot: { issueId: conversation.issueId, source: "chat:slack", endpointId: endpoint.id, wakeCommentId: link.commentId },
});
await db.update(chatConversations).set({ state: "unavailable" }).where(eq(chatConversations.id, conversation.id));
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
const notice = () => db.select().from(chatPublications).where(and(eq(chatPublications.issueId, conversation.issueId),
eq(chatPublications.idempotencyKey, `run:${runId}:failed:${endpoint.id}`)));
try {
await chatRunMilestoneWork(db).register(coordinator, () => true).ready;
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull());
expect(await notice()).toEqual([]);
const wakesBefore = wakeup.mock.calls.length;
await send("4100.2");
expect(wakeup.mock.calls.length).toBe(wakesBefore + 1);
// The harness accepts the new wake without creating a run. Existing
// projection debt must wake from ingress, not a later run transition.
await expect.poll(notice, { timeout: 5000 }).toEqual([
expect.objectContaining({ idempotencyKey: `run:${runId}:failed:${endpoint.id}` }),
]);
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull());
expect(onError).not.toHaveBeenCalled();
} finally { await coordinator.stop(); }
});
it.each(["milliseconds", "microseconds"])("drains bounded Slack chat-origin milestones without admitting an internal issue run (%s)", async (precision) => {
const fixture = await seedCompany();
const { callbacks, endpoint } = await configuredSlackEndpoint(fixture);
+5
View File
@@ -1,6 +1,7 @@
import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js";
import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js";
import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js";
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
import { notifyChatEndpointConfiguration } from "./chat-channels/configuration-work-notifications.js";
import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js";
import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js";
@@ -2125,6 +2126,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
taskUserId: string | null,
) => {
let conversation = existingConversation;
if (conversation) {
await notifyChatRunMilestones(taskTx, { companyId: taskEndpoint.companyId, issueId: conversation.issueId });
}
if (!conversation) {
const [taskResource] = await taskTx.select({ executionDefaults: chatEndpointResources.executionDefaults })
.from(chatEndpointResources).where(and(eq(chatEndpointResources.id, resource.id), eq(chatEndpointResources.endpointId, taskEndpoint.id)));
@@ -2166,6 +2170,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
},
taskTx,
);
await notifyChatRunMilestones(taskTx, { companyId: taskEndpoint.companyId, issueId: issue.id });
await taskTx
.insert(chatConversations)
.values({