Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-06 05:56:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix completed-history owner receipts and verified attachment cleanup

Browse Source 
Use truthful storage absence receipts for MCP deletion and preserve replay fences. Qualify retained chat controls and provider submissions through their actual owners, retain voice and registration obligations, and bound delivery receipt lookup.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-11-02 14:12:41 -06:00
1 parent aaf55692c4
commit 56acdfdffa
31 files changed
+1849 -98
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/services/chat-channels/discord/idle-history-proof.ts
+25
View File
@@ -0,0 +1,25 @@
/** Command acceptance is final history; a staged control publication remains an
* independent obligation and must settle before its command receipt is quiet. */
export function completedDiscordAction(a: string): string {
return `${a}.kind = 'discord_native_command' AND ${a}.status = 'processed'
AND ${a}.payload->>'version' = '1' AND ${a}.payload->'invocation'->>'schema' = 'paperclip.discord.native-command.v1'
AND ${a}.payload->'invocation'->>'command' IN ('status', 'new', 'close')
AND ${a}.payload->'invocation'->>'digest' ~ '^[a-f0-9]{64}$'
AND ${a}.provider_action_id = 'discord-native-command:' || (${a}.payload->'invocation'->>'interactionId')
AND ${a}.result->>'kind' = 'discord_native_command_recorded' AND length(${a}.result->>'content') > 0
AND EXISTS (SELECT 1 FROM chat_external_principals principal WHERE principal.id = ${a}.principal_id
AND principal.company_id = ${a}.company_id AND principal.provider = 'discord'
AND principal.external_id = ${a}.payload->'invocation'->>'actorExternalId')
AND ((${a}.conversation_id IS NULL AND ${a}.payload->'target' = 'null'::jsonb)
OR EXISTS (SELECT 1 FROM chat_conversations conversation WHERE conversation.id = ${a}.conversation_id
AND conversation.company_id = ${a}.company_id AND conversation.endpoint_id = ${a}.endpoint_id
AND conversation.id::text = ${a}.payload->'target'->>'conversationId'
AND conversation.issue_id::text = ${a}.payload->'target'->>'issueId'
AND conversation.session_generation::text = ${a}.payload->'target'->>'sessionGeneration'))
AND (${a}.result->'publicationId' = 'null'::jsonb OR EXISTS (
SELECT 1 FROM chat_publications publication WHERE publication.id::text = ${a}.result->>'publicationId'
AND publication.company_id = ${a}.company_id AND publication.endpoint_id = ${a}.endpoint_id
AND publication.conversation_id = ${a}.conversation_id
AND ((publication.state = 'published' AND publication.published_at IS NOT NULL)
OR (publication.state IN ('failed', 'cancelled') AND publication.next_attempt_at IS NULL))))`;
}
server/src/services/chat-channels/github/idle-history-proof.ts
+63 -7
View File
@@ -1,9 +1,65 @@
function completedIngress(a: string): string {
return `(${a}.status = 'processed' AND ${a}.result->>'httpStatus' ~ '^2[0-9][0-9]$' AND ${a}.result->>'retryable' = 'false')
OR (${a}.status = 'failed' AND ${a}.result->>'retryable' = 'false' AND (
${a}.result->>'code' = 'github_webhook_ingress_payload_invalid'
OR ${a}.result->>'httpStatus' ~ '^4[0-9][0-9]$'
OR (jsonb_typeof(${a}.result->'attempts') = 'number' AND (${a}.result->>'attempts')::numeric >= 5)))
OR (${a}.status = 'cancelled' AND ${a}.result->>'retryable' = 'false'
AND ${a}.result->>'code' IN ('github_webhook_ingress_runtime_superseded', 'github_webhook_recovery_source_changed'))`;
}
export function completedGithubAction(a: string): string {
return `(${a}.kind = 'github_response_comment' AND ${a}.status = 'processed' AND ${a}.result->>'final' = 'true')
OR (${a}.kind = 'github_webhook_ingress' AND (
(${a}.status = 'processed' AND ${a}.result->>'httpStatus' = '202' AND ${a}.result->>'retryable' = 'false')
OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'github_webhook_ingress_payload_invalid'
AND ${a}.result->>'retryable' = 'false')))
OR (${a}.kind = 'github_review_check' AND ${a}.status = 'processed'
AND ${a}.result->>'retryable' = 'false')`;
return `(${a}.kind = 'github_omission_navigation' AND ${a}.status = 'processed' AND ${a}.result IS NULL
AND ${a}.payload->>'version' = '1' AND ${a}.payload->>'preparedTextSha256' ~ '^[a-f0-9]{64}$'
AND ${a}.provider_action_id = 'github-omission-navigation:' || (${a}.payload->>'publicationId')
AND EXISTS (SELECT 1 FROM chat_publications publication JOIN heartbeat_runs run
ON run.id::text = ${a}.payload->>'runId' AND run.company_id = publication.company_id
AND run.native_issue_id = publication.issue_id AND run.runtime_mode = 'native'
JOIN native_run_finalizations finalization ON finalization.run_id = run.id AND finalization.company_id = run.company_id
AND finalization.issue_id = run.native_issue_id AND finalization.result_id::text = ${a}.payload->>'resultId' AND finalization.phase = 'committed'
WHERE publication.id::text = ${a}.payload->>'publicationId' AND publication.company_id = ${a}.company_id
AND publication.endpoint_id = ${a}.endpoint_id AND publication.conversation_id = ${a}.conversation_id
AND run.status = 'succeeded' AND run.finished_at IS NOT NULL AND run.cost_accounting_pending = false
AND ((publication.state = 'published' AND publication.published_at IS NOT NULL)
OR (publication.state IN ('failed', 'cancelled') AND publication.next_attempt_at IS NULL))))
OR (${a}.kind = 'github_response_comment' AND ${a}.result->>'final' = 'true'
AND (${a}.status = 'processed' OR (${a}.status = 'cancelled' AND ${a}.result->>'code'
IN ('run_already_answered', 'passive_response_wait', 'response_deleted'))))
OR (${a}.kind = 'github_pr_response_comment' AND ${a}.status = 'processed'
AND (${a}.result IS NULL OR (${a}.result->>'final' = 'true' AND length(${a}.result->>'id') > 0
AND length(${a}.result->>'url') > 0))
AND EXISTS (SELECT 1 FROM chat_conversations conversation WHERE conversation.id = ${a}.conversation_id
AND conversation.company_id = ${a}.company_id AND conversation.endpoint_id = ${a}.endpoint_id)
AND NOT EXISTS (SELECT 1 FROM chat_actions request WHERE request.kind = 'github_response_comment'
AND request.company_id = ${a}.company_id AND request.endpoint_id = ${a}.endpoint_id
AND request.conversation_id = ${a}.conversation_id AND (request.result->>'final' = 'true') IS NOT TRUE))
OR (${a}.kind = 'github_webhook_ingress' AND (${completedIngress(a)}))
OR (${a}.kind = 'github_review_check' AND (
(${a}.status = 'processed' AND (${a}.result->>'state' IN ('manual_required', 'incomplete', 'error')
OR (${a}.result->>'code' = 'assessment_owns_check' AND EXISTS (
SELECT 1 FROM chat_github_reviews review WHERE review.id::text = ${a}.result->>'reviewId'
AND review.company_id = ${a}.company_id AND review.endpoint_id = ${a}.endpoint_id AND review.assessment IS NOT NULL))))
OR (${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('connection_changed', 'stale_head')
AND (${a}.result->>'code' <> 'stale_head' OR NOT EXISTS (
SELECT 1 FROM chat_github_reviews review JOIN chat_deliveries delivery
ON delivery.id::text = review.delivery_id AND delivery.endpoint_id = review.endpoint_id
WHERE review.company_id = ${a}.company_id AND review.endpoint_id = ${a}.endpoint_id
AND review.state IN ('queued', 'running') AND review.assessment IS NULL
AND delivery.normalized_event->'githubAutomatic'->'context'->>'deliveryId' = ${a}.payload->'event'->>'deliveryId')))))
OR (${a}.kind = 'github_review_publication' AND (
(${a}.status = 'processed' AND (
(${a}.payload->>'operation' IN ('comment', 'update_comment', 'formal_review')
AND length(${a}.result->>'id') > 0 AND length(${a}.result->>'url') > 0)
OR (${a}.payload->>'operation' = 'assessment' AND EXISTS (SELECT 1 FROM chat_github_reviews review
WHERE review.id::text = ${a}.payload->>'reviewId' AND review.id::text = ${a}.result->>'reviewId'
AND review.company_id = ${a}.company_id AND review.endpoint_id = ${a}.endpoint_id
AND review.assessment IS NOT NULL AND review.check_url = ${a}.result->>'checkUrl'))))
OR (${a}.status = 'cancelled' AND ${a}.result->>'retryable' = 'false'
AND ${a}.result->>'code' IN ('publication_binding_unavailable', 'stale_head', 'authorization_changed'))))
OR (${a}.kind = 'github_webhook_recovery' AND (
(${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('not_eligible_for_recovery', 'source_changed_or_unavailable'))
OR (${a}.status = 'processed' AND EXISTS (SELECT 1 FROM chat_actions ingress
WHERE ingress.kind = 'github_webhook_ingress' AND ingress.company_id = ${a}.company_id
AND ingress.endpoint_id = ${a}.endpoint_id AND ingress.payload->>'deliveryId' = ${a}.payload->'original'->>'guid'
AND (${completedIngress('ingress')})))))`;
}
server/src/services/chat-channels/idle-history-proof.ts
+51 -16
View File
@@ -1,6 +1,11 @@
import { finishedIdleRun, finishedIdleWake } from "../idle-work-proof.js";
import { completedGithubAction } from "./github/idle-history-proof.js";
import { completedSlackAction } from "./slack/idle-history-proof.js";
import { completedExpiredChatControl, completedQuestionFormAction } from "../chat-question-form-idle-proof.js";
import { completedPhotonAction } from "./imessage-photon/idle-history-proof.js";
import { completedDiscordAction } from "./discord/idle-history-proof.js";
import { completedVoiceAction } from "../voice/idle-history-proof.js";
import { completedTeamsAction } from "./microsoft-teams/idle-history-proof.js";
import { completedTelegramAction } from "./telegram/idle-history-proof.js";
/** A processed inbound receipt can still own an unconsumed wake. */
@@ -8,12 +13,18 @@ function completedAction(a: string): string {
const wake = `EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${a}.company_id
AND w.id = ${a}.id AND w.agent_id::text = ${a}.payload->>'agentId'
AND w.payload->>'issueId' = ${a}.payload->>'issueId' AND ${finishedIdleWake('w')})`;
return `jsonb_typeof(${a}.payload) = 'object' AND jsonb_typeof(${a}.result) = 'object'
return `jsonb_typeof(${a}.payload) = 'object'
AND EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = ${a}.endpoint_id AND e.company_id = ${a}.company_id)
AND (${a}.delivery_id IS NULL OR EXISTS (SELECT 1 FROM chat_deliveries d
WHERE d.id = ${a}.delivery_id AND d.company_id = ${a}.company_id AND d.endpoint_id = ${a}.endpoint_id))
AND CASE
WHEN ${a}.kind = 'inbound_wakeup' THEN ${a}.status IN ('processed', 'failed') AND ${wake}
WHEN ${a}.kind = 'inbound_wakeup' THEN (${a}.status IN ('processed', 'failed') AND ${wake})
OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'inbound_wakeup_delivery_rejected'
AND EXISTS (SELECT 1 FROM chat_deliveries rejected WHERE rejected.id = ${a}.delivery_id
AND rejected.company_id = ${a}.company_id AND rejected.endpoint_id = ${a}.endpoint_id
AND rejected.state IN ('filtered', 'failed') AND rejected.next_attempt_at IS NULL))
WHEN ${a}.kind IN ('question_form_open', 'question_form_submit', 'question_form_open_attempt', 'question_answer')
THEN (${completedQuestionFormAction(a)})
WHEN ${a}.kind = 'interaction_wakeup' THEN ${a}.status = 'processed' AND
(${a}.result->>'code' = 'interaction_wakeup_no_longer_applicable' OR EXISTS (
SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${a}.company_id
@@ -28,40 +39,64 @@ function completedAction(a: string): string {
WHEN ${a}.kind = 'receipt_reaction' THEN
(${a}.status = 'processed' AND jsonb_typeof(${a}.result->'attempts') = 'number' AND (${a}.result->>'attempts')::numeric > 0)
OR (${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('receipt_reaction_superseded_by_terminal_publication', 'receipt_reaction_runtime_superseded'))
WHEN ${a}.kind = 'confirmation_response' THEN ${a}.status = 'processed'
WHEN ${a}.kind = 'confirmation_response' THEN (${completedExpiredChatControl(a)}) OR (${a}.status = 'processed'
AND ${a}.result->>'code' IN ('confirmation_resolution_committed_by_provider', 'confirmation_resolution_reconciled')
AND EXISTS (SELECT 1 FROM issue_thread_interactions interaction WHERE interaction.id::text = ${a}.result->>'interactionId'
AND interaction.company_id = ${a}.company_id AND interaction.status = ${a}.result->>'interactionStatus'
AND interaction.resolved_at IS NOT NULL)
AND interaction.resolved_at IS NOT NULL))
WHEN ${a}.kind = 'board_send_rejected' THEN ${a}.status = 'processed'
AND ${a}.result->>'code' = 'chat_board_send_attachments_already_bound'
AND ${a}.payload->>'schemaVersion' = '1' AND length(${a}.payload->>'requestedByUserId') > 0
AND jsonb_typeof(${a}.payload->'attachmentIds') = 'array' AND jsonb_array_length(CASE WHEN jsonb_typeof(${a}.payload->'attachmentIds') = 'array' THEN ${a}.payload->'attachmentIds' ELSE '[]'::jsonb END) BETWEEN 1 AND 20
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(CASE WHEN jsonb_typeof(${a}.payload->'attachmentIds') = 'array' THEN ${a}.payload->'attachmentIds' ELSE '[]'::jsonb END) rejected_id(value)
WHERE rejected_id.value !~ '^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$')
AND EXISTS (SELECT 1 FROM chat_conversations conversation WHERE conversation.id = ${a}.conversation_id
AND conversation.company_id = ${a}.company_id AND conversation.endpoint_id = ${a}.endpoint_id
AND conversation.issue_id::text = ${a}.payload->>'issueId'
AND ${a}.provider_action_id = 'board-send-rejected:' || conversation.id::text || ':' || (${a}.payload->>'idempotencyKey'))
WHEN ${a}.kind = 'task_control_authorization' THEN ${a}.status IN ('processed', 'cancelled')
AND EXISTS (SELECT 1 FROM chat_publications p WHERE p.company_id = ${a}.company_id AND p.endpoint_id = ${a}.endpoint_id
AND p.conversation_id = ${a}.conversation_id AND ${a}.provider_action_id = 'task-control-authorization:' || p.id::text
AND ((p.state = 'published' AND p.published_at IS NOT NULL) OR p.state = 'cancelled'))
ELSE (${[completedGithubAction, completedSlackAction, completedTelegramAction]
ELSE (${[completedGithubAction, completedSlackAction, completedTelegramAction, completedTeamsAction, completedPhotonAction, completedDiscordAction, completedVoiceAction]
.map(proof => `(${proof(a)})`).join(' OR ')}) END`;
}
export const idleChatHistoryChecks = [
{ reason: "chat_deliveries", query: `WITH action_outcomes AS MATERIALIZED (
SELECT a.delivery_id, a.kind, (${completedAction('a')}) IS TRUE AS completed
FROM chat_actions a WHERE a.delivery_id IS NOT NULL
), action_receipts AS (
SELECT delivery_id, bool_and(completed) AS completed,
bool_or(kind = 'inbound_wakeup' AND completed) AS consumed
FROM action_outcomes GROUP BY delivery_id
) SELECT 1 FROM chat_deliveries d LEFT JOIN action_receipts receipt ON receipt.delivery_id = d.id WHERE (
{ reason: "chat_deliveries", query: `WITH candidate_deliveries AS MATERIALIZED (
SELECT d.id, d.company_id, d.endpoint_id, d.state, d.event_kind FROM chat_deliveries d WHERE (
EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = d.endpoint_id AND e.company_id = d.company_id)
AND jsonb_typeof(d.normalized_event) = 'object'
AND d.normalized_event->>'providerEventId' = d.provider_event_id
AND d.normalized_event->>'kind' = d.event_kind
AND d.event_kind IN ('mention', 'message', 'direct_message', 'message_updated', 'message_deleted', 'message_restored',
'reaction_added', 'reaction_removed', 'action', 'modal_submitted', 'modal_closed', 'slash_command', 'file_shared', 'installation', 'uninstallation', 'unknown')
AND (d.state <> 'processed' OR d.event_kind NOT IN ('mention', 'message', 'direct_message')
OR receipt.consumed IS TRUE)
AND d.next_attempt_at IS NULL AND (d.state = 'failed' OR (d.state IN ('processed', 'filtered') AND d.processed_at IS NOT NULL))
AND (d.conversation_id IS NULL OR EXISTS (SELECT 1 FROM chat_conversations c WHERE c.id = d.conversation_id
AND c.company_id = d.company_id AND c.endpoint_id = d.endpoint_id))
AND COALESCE(receipt.completed, true)) IS NOT TRUE` },
) IS TRUE
), action_outcomes AS MATERIALIZED (
SELECT a.delivery_id, a.kind, (${completedAction('a')}) IS TRUE AS completed
FROM chat_actions a JOIN candidate_deliveries candidate ON candidate.id = a.delivery_id
), action_receipts AS MATERIALIZED (
SELECT delivery_id, bool_and(completed) AS completed,
bool_or(kind = 'inbound_wakeup' AND completed) AS consumed
FROM action_outcomes GROUP BY delivery_id
) SELECT 1 FROM chat_deliveries d WHERE (
EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = d.endpoint_id AND e.company_id = d.company_id)
AND jsonb_typeof(d.normalized_event) = 'object'
AND d.normalized_event->>'providerEventId' = d.provider_event_id
AND d.normalized_event->>'kind' = d.event_kind
AND d.event_kind IN ('mention', 'message', 'direct_message', 'message_updated', 'message_deleted', 'message_restored',
'reaction_added', 'reaction_removed', 'action', 'modal_submitted', 'modal_closed', 'slash_command', 'file_shared', 'installation', 'uninstallation', 'unknown')
AND d.next_attempt_at IS NULL AND (d.state = 'failed' OR (d.state IN ('processed', 'filtered') AND d.processed_at IS NOT NULL))
AND (d.conversation_id IS NULL OR EXISTS (SELECT 1 FROM chat_conversations c WHERE c.id = d.conversation_id
AND c.company_id = d.company_id AND c.endpoint_id = d.endpoint_id))
) IS NOT TRUE
UNION ALL SELECT 1 FROM candidate_deliveries d
WHERE d.id IN (SELECT receipt.delivery_id FROM action_receipts receipt WHERE NOT receipt.completed)
OR (d.state = 'processed' AND d.event_kind IN ('mention', 'message', 'direct_message')
AND d.id NOT IN (SELECT receipt.delivery_id FROM action_receipts receipt WHERE receipt.consumed))` },
{ reason: "chat_publications", query: `SELECT 1 FROM chat_publications p WHERE (
(p.state = 'published' AND p.published_at IS NOT NULL AND p.redacted_error IS NULL
OR (p.state IN ('failed', 'cancelled') AND p.next_attempt_at IS NULL))
server/src/services/chat-channels/imessage-photon/events.ts
+4 -37
View File
@@ -1,11 +1,10 @@
import { stagePhotonResponseNotice } from "./response-notices.js";
import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js";
import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
import { nativeSha256 } from "../../native-runtime/canonical.js";
import { projectSafeChatPublicationText } from "../../chat-publication-projection.js";
import { PhotonAnswerValidationError, nativePhotonInteraction, photonResponseCommand, parsePhotonQuestionAnswer, type PhotonPromptReceipt, type PhotonInteractionBinding, type PhotonDraft } from "../../photon/interactions.js";
import { validateNativeQuestionResponseInput } from "../../native-runtime/native-question-bridge.js";
import type { AskUserQuestionsAnswer, AskUserQuestionsInteraction, IssueThreadInteraction } from "@paperclipai/shared";
@@ -368,41 +367,9 @@ export function createPhotonEvents(db: Db, options: PhotonEventsOptions) {
}
const enqueueResponse = async (key: string, text: string) => {
await assertCurrent();
await db.transaction(async (tx) => {
await tx
.insert(chatActions)
.values({
companyId: endpoint.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
kind: "photon_response_notice",
providerActionId: `photon-notice:${key}`,
payload: {
reference,
questionIndex,
publicationId: binding.publicationId,
},
status: "processed",
})
.onConflictDoNothing();
await notifyChatPublicationWork(tx);
await tx
.insert(chatPublications)
.values({
companyId: endpoint.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
issueId: conversation.issueId,
idempotencyKey: key,
payload: {
...publication.payload,
card: undefined,
text: projectSafeChatPublicationText(text),
},
state: "pending",
})
.onConflictDoNothing();
});
await db.transaction(tx => stagePhotonResponseNotice(tx, {
publication, reference, questionIndex, key, text,
}));
scheduleMessageProcessing(() =>
processPendingPublications().then(() => undefined),
);
server/src/services/chat-channels/imessage-photon/idle-history-proof.ts
+58
View File
@@ -0,0 +1,58 @@
/** Photon tokens and immutable provenance receipts do not own a turn once their
* exact canonical owner has settled. Receiver ownership remains a separate gate. */
export function completedPhotonAction(a: string): string {
const publication = (id: string) => `EXISTS (SELECT 1 FROM chat_publications publication
WHERE publication.id::text = ${id} AND publication.company_id = ${a}.company_id
AND publication.endpoint_id = ${a}.endpoint_id AND publication.conversation_id = ${a}.conversation_id
AND ((publication.state = 'published' AND publication.published_at IS NOT NULL)
OR (publication.state IN ('failed', 'cancelled') AND publication.next_attempt_at IS NULL)))`;
return `(${a}.kind = 'photon_interaction' AND (${a}.status IN ('issued', 'processed') OR (${a}.status = 'expired' AND ${a}.result->>'code' = 'interaction_resolved_elsewhere'))
AND ${a}.payload->>'version' = '1' AND length(${a}.payload->>'reference') > 0
AND ${a}.provider_action_id = 'photon:' || (${a}.payload->>'reference')
AND EXISTS (SELECT 1 FROM issue_thread_interactions interaction JOIN chat_publications publication
ON publication.id::text = ${a}.payload->>'publicationId' AND publication.company_id = interaction.company_id
AND publication.issue_id = interaction.issue_id AND publication.payload->>'interactionId' = interaction.id::text
JOIN chat_conversations conversation ON conversation.id = publication.conversation_id
AND conversation.company_id = interaction.company_id AND conversation.issue_id = interaction.issue_id
WHERE interaction.id::text = ${a}.payload->>'interactionId' AND interaction.company_id = ${a}.company_id
AND interaction.kind IN ('ask_user_questions', 'request_confirmation')
AND interaction.status IN ('answered', 'accepted', 'rejected', 'cancelled', 'expired') AND interaction.resolved_at IS NOT NULL
AND publication.endpoint_id = ${a}.endpoint_id AND conversation.id = ${a}.conversation_id
AND conversation.session_generation::text = ${a}.payload->>'sessionGeneration'))
OR (${a}.kind = 'photon_publication_retry' AND ${a}.status = 'processed'
AND length(${a}.result->>'authorizedByUserId') > 0 AND ${a}.payload->>'attempt' ~ '^[1-9][0-9]*$'
AND ${a}.provider_action_id = 'photon-retry:' || (${a}.payload->>'publicationId') || ':' || (${a}.payload->>'attempt')
AND ${publication(`${a}.payload->>'publicationId'`)})
OR (${a}.kind = 'photon_response_notice' AND ${a}.status = 'processed' AND ${a}.result IS NULL
AND length(${a}.payload->>'reference') > 0 AND ${a}.payload->>'questionIndex' ~ '^[0-9]+$'
AND EXISTS (SELECT 1 FROM chat_publications notice WHERE notice.company_id = ${a}.company_id
AND notice.endpoint_id = ${a}.endpoint_id AND notice.conversation_id = ${a}.conversation_id
AND ${a}.provider_action_id = 'photon-notice:' || notice.idempotency_key
AND ((notice.state = 'published' AND notice.published_at IS NOT NULL)
OR (notice.state IN ('failed', 'cancelled') AND notice.next_attempt_at IS NULL)))
AND EXISTS (SELECT 1 FROM chat_publications source WHERE source.id::text = ${a}.payload->>'publicationId'
AND source.company_id = ${a}.company_id AND source.endpoint_id = ${a}.endpoint_id
AND source.conversation_id = ${a}.conversation_id))
OR (${a}.kind IN ('attachment_derivative', 'attachment_companion') AND ${a}.status = 'processed' AND ${a}.result IS NULL
AND EXISTS (SELECT 1 FROM chat_endpoints endpoint WHERE endpoint.id = ${a}.endpoint_id
AND endpoint.company_id = ${a}.company_id AND endpoint.provider = 'imessage-photon')
AND ${a}.delivery_id IS NOT NULL
AND ((${a}.kind = 'attachment_derivative' AND ${a}.payload->>'kind' = 'heif_jpeg_preview')
OR (${a}.kind = 'attachment_companion' AND ${a}.payload->>'kind' = 'live_photo_video'))
AND ${a}.payload->>'originalAttachmentId' ~ '^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$'
AND ${a}.payload->>'derivativeAttachmentId' ~ '^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$'
AND ${a}.payload->>'originalAttachmentId' <> ${a}.payload->>'derivativeAttachmentId'
AND ${a}.payload->>'originalSha256' ~ '^[a-f0-9]{64}$' AND ${a}.payload->>'derivativeSha256' ~ '^[a-f0-9]{64}$'
AND ${a}.provider_action_id = 'photon-' || (${a}.payload->>'kind') || ':' || ${a}.delivery_id::text || ':' || (${a}.payload->>'originalAttachmentId')
AND NOT EXISTS (SELECT 1 FROM issue_attachments attachment WHERE attachment.id::text IN
(${a}.payload->>'originalAttachmentId', ${a}.payload->>'derivativeAttachmentId')
AND (attachment.company_id <> ${a}.company_id OR NOT EXISTS (SELECT 1 FROM assets asset
WHERE asset.id = attachment.asset_id AND asset.company_id = ${a}.company_id AND asset.sha256 = CASE
WHEN attachment.id::text = ${a}.payload->>'originalAttachmentId' THEN ${a}.payload->>'originalSha256'
ELSE ${a}.payload->>'derivativeSha256' END)))
AND NOT EXISTS (SELECT 1 FROM issue_attachments original JOIN issue_attachments derivative
ON derivative.id::text = ${a}.payload->>'derivativeAttachmentId'
WHERE original.id::text = ${a}.payload->>'originalAttachmentId'
AND (original.company_id <> derivative.company_id OR original.issue_id <> derivative.issue_id
OR original.issue_comment_id IS DISTINCT FROM derivative.issue_comment_id)))`;
}
server/src/services/chat-channels/imessage-photon/response-notices.ts
+27
View File
@@ -0,0 +1,27 @@
import { chatActions, chatPublications, type Db } from "@paperclipai/db";
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
import { projectSafeChatPublicationText } from "../../chat-publication-projection.js";
type Tx = Parameters<Parameters<Db["transaction"]>[0]>[0];
/** Keep the notice marker and its outbox row in the same owner transaction. */
export async function stagePhotonResponseNotice(tx: Tx, input: {
publication: typeof chatPublications.$inferSelect;
reference: string;
questionIndex: number;
key: string;
text: string;
}) {
const { publication, reference, questionIndex, key, text } = input;
await tx.insert(chatActions).values({
companyId: publication.companyId, endpointId: publication.endpointId, conversationId: publication.conversationId,
kind: "photon_response_notice", providerActionId: `photon-notice:${key}`,
payload: { reference, questionIndex, publicationId: publication.id }, status: "processed",
}).onConflictDoNothing();
await notifyChatPublicationWork(tx);
await tx.insert(chatPublications).values({
companyId: publication.companyId, endpointId: publication.endpointId, conversationId: publication.conversationId,
issueId: publication.issueId, idempotencyKey: key,
payload: { ...publication.payload, card: undefined, text: projectSafeChatPublicationText(text) }, state: "pending",
}).onConflictDoNothing();
}
server/src/services/chat-channels/microsoft-teams/idle-history-proof.ts
+35
View File
@@ -0,0 +1,35 @@
/** Teams keeps immutable provider-effect and recipient-intent receipts after
* delivery. Only the exact closed transfer/publication retires that authority. */
export function completedTeamsAction(a: string): string {
return `${a}.status = 'processed' AND ${a}.kind IN ('teams_file_effect_intent', 'teams_file_consent', 'teams_board_file_intent')
AND EXISTS (SELECT 1 FROM chat_teams_file_transfers transfer JOIN chat_publications publication
ON publication.id = transfer.publication_id AND publication.company_id = transfer.company_id
WHERE transfer.company_id = ${a}.company_id AND transfer.endpoint_id = ${a}.endpoint_id
AND transfer.conversation_id = ${a}.conversation_id
AND publication.id::text = ${a}.payload->>'publicationId'
AND publication.endpoint_id = ${a}.endpoint_id AND publication.conversation_id = ${a}.conversation_id
AND transfer.phase IN ('delivered', 'declined', 'expired', 'cancelled')
AND transfer.attempt_id IS NULL AND transfer.attempt_expires_at IS NULL
AND ((publication.state = 'published' AND publication.published_at IS NOT NULL)
OR (publication.state IN ('failed', 'cancelled') AND publication.next_attempt_at IS NULL))
AND CASE WHEN ${a}.kind = 'teams_file_effect_intent' THEN
${a}.payload->>'schema' = 'paperclip.teams.file-effect-intent.v1'
AND ${a}.payload->>'transferId' = transfer.id::text
AND ${a}.provider_action_id = 'teams-file-effect:' || (${a}.payload->>'attemptId')
AND ${a}.payload->>'stage' IN ('consent', 'upload', 'file_info')
WHEN ${a}.kind = 'teams_file_consent' THEN
${a}.payload->>'schema' = 'paperclip.teams.consent-receipt.v1'
AND ${a}.payload->>'transferId' = transfer.id::text
AND ${a}.payload->>'receiptDigest' = transfer.response_digest
AND ${a}.payload->>'action' IN ('accept', 'decline')
AND ${a}.principal_id IS NOT DISTINCT FROM transfer.principal_id
ELSE ${a}.payload->>'schema' = 'paperclip.teams.board-file-intent.v1'
AND ${a}.provider_action_id = 'teams_board_file_intent:' || publication.id::text
AND ${a}.payload->>'issueId' = transfer.issue_id::text
AND ${a}.payload->>'attachmentId' = transfer.attachment_id::text
AND ${a}.payload->>'commentId' = publication.comment_id::text
AND ${a}.payload->>'sourceScopeSha256' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'recipientBindingSha256' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'byteIdentitySha256' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'commentBodySha256' ~ '^[a-f0-9]{64}$' END)`;
}
server/src/services/chat-channels/slack/idle-history-proof.ts
+62 -1
View File
@@ -1,3 +1,63 @@
/** Connector rows below are immutable authority/confirmed-call receipts. Their
* own downstream publication, interaction and run rows keep independent gates. */
function completedSlackConnectorAction(a: string): string {
const principal = `EXISTS (SELECT 1 FROM chat_external_principals principal WHERE principal.id = ${a}.principal_id
AND principal.company_id = ${a}.company_id AND principal.provider = 'slack')`;
const issue = `EXISTS (SELECT 1 FROM issues issue WHERE issue.id::text = ${a}.payload->>'issueId'
AND issue.company_id = ${a}.company_id AND (${a}.conversation_id IS NULL OR EXISTS (
SELECT 1 FROM chat_conversations conversation WHERE conversation.id = ${a}.conversation_id
AND conversation.company_id = issue.company_id AND conversation.issue_id = issue.id
AND conversation.endpoint_id = ${a}.endpoint_id)))`;
return `(${a}.kind = 'slack_private_source' AND ${a}.status = 'processed' AND ${a}.result IS NULL
AND ${issue} AND ${principal} AND length(${a}.payload->>'channelId') > 0
AND length(${a}.payload->>'requesterId') > 0
AND ${a}.provider_action_id = 'slack-private:' || (${a}.payload->>'issueId') || ':'
|| (${a}.payload->>'channelId') || ':' || (${a}.payload->>'requesterId'))
OR (${a}.kind = 'slack_connect' AND ${a}.status = 'processed'
AND ${a}.payload->>'version' = '1' AND ${a}.result->>'code' = 'slack_identity_discovered'
AND length(${a}.payload->>'channelId') > 0 AND ${principal}
AND EXISTS (SELECT 1 FROM chat_external_principals principal WHERE principal.id = ${a}.principal_id
AND principal.external_id = ${a}.payload->>'userId'))
OR (${a}.kind IN ('slack_test_message', 'slack_test_message_invitation') AND ${a}.status = 'processed'
AND length(${a}.payload->>'userId') > 0 AND ${a}.payload->>'slackUserId' ~ '^[UW][A-Z0-9]+$'
AND CASE WHEN ${a}.kind = 'slack_test_message' THEN ${a}.result->>'slackUrl' ~ '^https://app[.]slack[.]com/client/[^/]+/D[A-Z0-9]+$'
ELSE ${principal} AND ${a}.result->>'expiresAt' ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T' END)
OR (${a}.kind = 'slack_managed_reconnect' AND ${a}.status = 'processed' AND ${a}.result IS NULL
AND ${a}.provider_action_id LIKE 'slack_managed_reconnect:%'
AND length(${a}.payload->>'registrationRequestId') > 0
AND EXISTS (SELECT 1 FROM chat_slack_manager_grants grant_record WHERE grant_record.id::text = ${a}.payload->>'grantId'
AND grant_record.company_id = ${a}.company_id))
OR (${a}.kind = 'slack_tool_write' AND ${a}.status = 'processed' AND jsonb_typeof(${a}.result) = 'object'
AND ${a}.payload->>'name' IN ('slack_open_dm', 'slack_post_message', 'slack_update_message', 'slack_delete_message',
'slack_add_reaction', 'slack_remove_reaction', 'slack_add_pin', 'slack_remove_pin', 'slack_add_bookmark',
'slack_remove_bookmark', 'slack_create_canvas', 'slack_edit_canvas', 'slack_replace_canvas_section',
'slack_share_list', 'slack_edit_list', 'slack_create_list', 'slack_create_list_item', 'slack_update_list_item',
'slack_invite', 'slack_create_channel', 'slack_upload_file')
AND ${a}.payload->>'hash' ~ '^[a-f0-9]{64}$' AND jsonb_typeof(${a}.payload->'args') = 'object'
AND length(${a}.payload->>'userId') > 0 AND length(${a}.payload->>'revision') > 0
AND ${a}.provider_action_id = 'slack-tool:' || (${a}.payload->'binding'->>'issueId') || ':' || (${a}.payload->'args'->>'idempotencyKey')
AND EXISTS (SELECT 1 FROM tool_invocations invocation JOIN heartbeat_runs run ON run.id = invocation.run_id
AND run.company_id = invocation.company_id AND run.agent_id = invocation.agent_id
WHERE invocation.id::text = ${a}.payload->>'invocationId' AND invocation.company_id = ${a}.company_id
AND invocation.run_id::text = ${a}.payload->'binding'->>'runId'
AND invocation.agent_id::text = ${a}.payload->'binding'->>'agentId'
AND invocation.issue_id::text = ${a}.payload->'binding'->>'issueId'
AND ${a}.payload->'binding'->>'companyId' = ${a}.company_id::text))
OR (${a}.kind = 'slack_board_message' AND ${issue} AND EXISTS (
SELECT 1 FROM issue_comments comment JOIN chat_publications publication
ON publication.comment_id = comment.id AND publication.company_id = comment.company_id
WHERE comment.id::text = ${a}.payload->>'commentId' AND comment.company_id = ${a}.company_id
AND comment.issue_id::text = ${a}.payload->>'issueId' AND comment.author_user_id = ${a}.payload->>'userId'
AND publication.endpoint_id = ${a}.endpoint_id AND publication.conversation_id = ${a}.conversation_id
AND ((publication.state = 'published' AND publication.published_at IS NOT NULL)
OR (publication.state IN ('failed', 'cancelled') AND publication.next_attempt_at IS NULL)))
AND ((${a}.status = 'cancelled' AND ${a}.result->>'code' = 'slack_board_work_not_authorized')
OR (${a}.status = 'processed' AND (${a}.result IS NULL OR (${a}.result->>'resumeApplied' = 'true'
AND EXISTS (SELECT 1 FROM agent_wakeup_requests wake WHERE wake.company_id = ${a}.company_id
AND wake.idempotency_key = 'slack-board-comment:' || ${a}.id::text AND wake.finished_at IS NOT NULL
AND wake.status IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out')))))))`;
}
export function completedSlackAction(a: string): string {
return `(${a}.kind = 'slack_session_sync' AND (
(${a}.status = 'processed' AND ${a}.result->>'outcome' IN ('updated', 'unsupported')
@@ -12,5 +72,6 @@ export function completedSlackAction(a: string): string {
AND ${a}.result->>'code' IN ('slack_session_stop_cancelled', 'slack_session_stop_target_superseded', 'slack_session_stop_no_current_work')
AND NOT EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id::text = ${a}.result->>'runId'
AND (r.company_id <> ${a}.company_id OR r.status NOT IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted')
OR r.finished_at IS NULL OR r.cost_accounting_pending)))`;
OR r.finished_at IS NULL OR r.cost_accounting_pending)))
OR (${completedSlackConnectorAction(a)})`;
}
server/src/services/chat-channels/telegram/idle-history-proof.ts
+9 -1
View File
@@ -1,5 +1,13 @@
export function completedTelegramAction(a: string): string {
return `(${a}.kind = 'telegram_maintenance' AND (
return `(${a}.kind = 'telegram_stop_subscription' AND ${a}.status = 'processed'
AND ${a}.result->>'code' = 'telegram_stop_subscription_confirmed' AND ${a}.payload->>'version' = '1'
AND ${a}.payload->>'botUserId' ~ '^[1-9][0-9]{0,15}$'
AND ${a}.payload->>'credentialFingerprint' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'webhookUrlSha256' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'runtimeGeneration' ~ '^[0-9]+$'
AND ${a}.provider_action_id = 'telegram-stop-subscription:' || (${a}.payload->>'runtimeGeneration') || ':'
|| (${a}.payload->>'credentialFingerprint') || ':' || (${a}.payload->>'webhookUrlSha256'))
OR (${a}.kind = 'telegram_maintenance' AND (
(${a}.status = 'processed' AND (${a}.result->>'providerConfirmed' = 'true'
OR ${a}.result->>'code' = 'telegram_stop_subscription_confirmed'))
OR (${a}.status = 'cancelled' AND ${a}.result->>'code' = 'telegram_maintenance_credentials_superseded')
server/src/services/chat-question-form-idle-proof.ts
+54
View File
@@ -0,0 +1,54 @@
/** Form-open tokens stay issued: resolving the canonical question revokes their
* authority. A processed submission additionally needs its exact answer receipt. */
export function completedQuestionFormAction(a: string): string {
const resolved = `EXISTS (SELECT 1 FROM issue_thread_interactions interaction
JOIN chat_conversations conversation ON conversation.issue_id = interaction.issue_id AND conversation.company_id = interaction.company_id
JOIN chat_publications publication ON publication.id::text = ${a}.payload->>'publicationId'
AND publication.company_id = interaction.company_id AND publication.issue_id = interaction.issue_id
AND publication.conversation_id = conversation.id AND publication.endpoint_id = conversation.endpoint_id
WHERE interaction.id::text = ${a}.payload->>'interactionId' AND interaction.company_id = ${a}.company_id
AND (interaction.kind = 'ask_user_questions' OR (interaction.kind = 'request_item_verdicts' AND EXISTS (
SELECT 1 FROM chat_endpoints endpoint WHERE endpoint.id = ${a}.endpoint_id AND endpoint.company_id = ${a}.company_id AND endpoint.provider = 'slack'))) AND interaction.status IN ('answered', 'cancelled', 'expired')
AND interaction.resolved_at IS NOT NULL AND conversation.id = ${a}.conversation_id
AND conversation.endpoint_id = ${a}.endpoint_id AND publication.payload->>'interactionId' = interaction.id::text)`;
return `(${a}.kind IN ('question_form_open', 'question_form_submit') AND ${a}.payload->>'version' = '1'
AND ${resolved} AND (
(${a}.kind = 'question_form_open' AND (${a}.status = 'issued' OR (${a}.status = 'expired'
AND ${a}.result->>'code' IN ('question_form_submitted', 'interaction_resolved_elsewhere'))) AND EXISTS (
SELECT 1 FROM chat_actions submit WHERE submit.company_id = ${a}.company_id AND submit.endpoint_id = ${a}.endpoint_id
AND submit.conversation_id = ${a}.conversation_id AND submit.kind = 'question_form_submit'
AND submit.provider_action_id = ${a}.payload->>'formActionId'
AND submit.payload->>'formActionId' = submit.provider_action_id
AND submit.payload->>'interactionId' = ${a}.payload->>'interactionId'
AND submit.payload->>'publicationId' = ${a}.payload->>'publicationId'
AND submit.payload->>'expiresAt' = ${a}.payload->>'expiresAt'
AND (submit.status = 'issued' OR (submit.status = 'expired' AND submit.result->>'code' = 'interaction_resolved_elsewhere') OR (submit.status = 'processed'
AND submit.result->>'code' = 'question_form_answered'
AND submit.result->>'interactionId' = submit.payload->>'interactionId'))))
OR (${a}.kind = 'question_form_submit' AND ${a}.payload->>'formActionId' = ${a}.provider_action_id
AND (${a}.status = 'issued' OR (${a}.status = 'expired' AND ${a}.result->>'code' = 'interaction_resolved_elsewhere') OR (${a}.status = 'processed' AND ${a}.result->>'code' = 'question_form_answered'
AND ${a}.result->>'interactionId' = ${a}.payload->>'interactionId')))))
OR (${a}.kind = 'question_answer' AND ${resolved}
AND (${a}.status = 'issued' OR (${a}.status = 'processed'
AND ${a}.result->>'interactionId' = ${a}.payload->>'interactionId')
OR (${a}.status = 'expired' AND ${a}.result->>'code' IN ('interaction_resolved_by_sibling_action', 'interaction_resolved_elsewhere'))))
OR (${a}.kind = 'question_form_open_attempt' AND ${a}.status = 'processed'
AND ${a}.result->>'code' = 'question_form_opened')`;
}
/** The terminal interaction writer expires its unused callback tokens after a
* board/sibling resolution. Expiry alone never proves canonical completion. */
export function completedExpiredChatControl(a: string): string {
return `${a}.kind = 'confirmation_response' AND ${a}.status = 'expired'
AND ${a}.result->>'code' = 'interaction_resolved_elsewhere' AND ${a}.payload->>'version' = '1'
AND ${a}.payload->>'decision' IN ('accept', 'reject') AND EXISTS (
SELECT 1 FROM issue_thread_interactions interaction JOIN chat_publications publication
ON publication.id::text = ${a}.payload->>'publicationId' AND publication.company_id = interaction.company_id
AND publication.issue_id = interaction.issue_id AND publication.payload->>'interactionId' = interaction.id::text
JOIN chat_conversations conversation ON conversation.id = publication.conversation_id
AND conversation.company_id = interaction.company_id AND conversation.issue_id = interaction.issue_id
WHERE interaction.id::text = ${a}.payload->>'interactionId' AND interaction.company_id = ${a}.company_id
AND interaction.kind IN ('request_confirmation', 'request_checkbox_confirmation', 'suggest_tasks', 'connection_intent')
AND interaction.status IN ('accepted', 'rejected', 'cancelled', 'expired') AND interaction.resolved_at IS NOT NULL
AND publication.endpoint_id = ${a}.endpoint_id AND conversation.id = ${a}.conversation_id)`;
}
server/src/services/idle-sleep-safety.ts
+1 -1
View File
@@ -17,7 +17,7 @@ import { idleBrowserHistoryChecks } from "./browser-use-idle-proof.js";
import { idleDecisionHistoryChecks } from "./decision-idle-proof.js";
import { idleWorkspaceHistoryChecks } from "./workspace-history-idle-proof.js";
import { idleMcpHistoryChecks } from "./public-mcp/idle-history-proof.js";
import { idleExternalAgentHistoryChecks } from "../modules/external-agents/idle-history-proof.js";
import { idleExternalAgentHistoryChecks } from "../modules/external-agents/index.js";
export const completedHistoryChecks = [
...idleChatHistoryChecks, ...idleToolActionHistoryChecks, ...idleBrowserHistoryChecks,
server/src/services/issues.ts
+7 -1
View File
@@ -1,3 +1,4 @@
import { recordMcpAttachmentCleanup, type McpAttachmentCleanup } from "./public-mcp/upload-cleanup.js";
import { notifyStrandedAgentStatus, notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
import { finishedIdleWake } from "./idle-work-proof.js";
import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
@@ -13907,7 +13908,7 @@ export function issueService(db: Db) {
.where(eq(issueAttachments.id, id))
.then((rows) => rows[0] ?? null),
removeAttachment: async (id: string) =>
removeAttachment: async (id: string, options?: { mcpCleanup?: McpAttachmentCleanup | null }) =>
db.transaction(async (tx) => {
const existing = await tx
.select({
@@ -13937,6 +13938,11 @@ export function issueService(db: Db) {
await notifyIssueRecovery(tx, { companyId: existing.companyId });
await tx.delete(issueAttachments).where(eq(issueAttachments.id, id));
await tx.delete(assets).where(eq(assets.id, existing.assetId));
if (options?.mcpCleanup?.id === existing.id && options.mcpCleanup.companyId === existing.companyId
&& options.mcpCleanup.issueId === existing.issueId && options.mcpCleanup.provider === existing.provider
&& options.mcpCleanup.objectKey === existing.objectKey) {
await recordMcpAttachmentCleanup(tx, options.mcpCleanup);
}
return existing;
}),
server/src/services/native-runtime/native-session-executor.test.ts
+5
View File
@@ -9897,6 +9897,11 @@ describe("native warm session supervision", () => {
it("blocks idle sleep for a released local warm session until its close checkpoint finishes", async () => {
await closeIdleWarmNativeSessionsForRestart();
const { reconcilePersistedRuntimeServicesOnStartup, readWorkspaceRuntimeIdleWork } = await import("../workspace-runtime.js");
// The real server reconstructs the workspace owner before startup can
// become quiet. This fixture has no runtime rows or desired services.
await reconcilePersistedRuntimeServicesOnStartup(leaseDb());
expect(readWorkspaceRuntimeIdleWork()).toBe("none");
markIdleIngressTracked(); markIdleStartupComplete({ scheduledBackups: false });
let finishClose!: () => void;
const checkpoint = new Promise<void>(resolve => { finishClose = resolve; });
server/src/services/public-mcp/file-transfers.ts
+18 -4
View File
@@ -80,13 +80,22 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
}
async function sweep() {
await db.delete(mcpFileTickets).where(lt(mcpFileTickets.expiresAt, new Date()));
// Serialize cleanup against completion/reissue. Never remove a committed file.
// Serialize cleanup against completion/reissue. Requalify old optimistic
// cleanup and orphaned removed attachments, never a still-linked file.
await db.transaction(async tx => {
const abandoned = await tx.select().from(mcpAttachmentUploads).where(and(isNull(mcpAttachmentUploads.attachmentId), isNull(mcpAttachmentUploads.cleanedAt), lt(mcpAttachmentUploads.expiresAt, new Date()))).limit(20).for("update", { skipLocked: true });
const abandoned = await tx.select().from(mcpAttachmentUploads).where(and(
isNull(mcpAttachmentUploads.cleanupVerifiedAt), lt(mcpAttachmentUploads.expiresAt, new Date()),
sql`(${mcpAttachmentUploads.attachmentId} IS NULL OR NOT EXISTS (
SELECT 1 FROM issue_attachments attachment WHERE attachment.id = ${mcpAttachmentUploads.attachmentId}))`,
)).limit(20).for("update", { skipLocked: true });
for (const upload of abandoned) {
if (upload.storageProvider !== storage.provider) continue;
await storage.deleteObject(upload.companyId, upload.objectKey);
await tx.update(mcpAttachmentUploads).set({ cleanedAt: new Date() }).where(eq(mcpAttachmentUploads.id, upload.id));
if ((await storage.headObject(upload.companyId, upload.objectKey)).exists !== false) {
throw new Error("MCP upload storage cleanup could not be confirmed");
}
const verifiedAt = new Date();
await tx.update(mcpAttachmentUploads).set({ cleanedAt: verifiedAt, cleanupVerifiedAt: verifiedAt }).where(eq(mcpAttachmentUploads.id, upload.id));
}
});
}
@@ -133,6 +142,7 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
const published = await db.transaction(async tx => {
const [upload] = await tx.select().from(mcpAttachmentUploads).where(eq(mcpAttachmentUploads.id, row.uploadId!)).for("update");
if (!upload || upload.companyId !== principal.grant.companyId || upload.userId !== principal.grant.userId || upload.workSessionId !== row.workSessionId || upload.grantId && upload.grantId !== row.grantId || upload.storageProvider !== storage.provider) throw unavailable();
if (upload.expiresAt <= new Date() || upload.cleanedAt || upload.cleanupVerifiedAt) throw unavailable();
if (upload.sha256 !== digest || upload.byteSize !== bytes.length || req.get("Content-Type")?.split(";")[0]?.trim().toLowerCase() !== upload.contentType) throw new McpCapabilityError("Upload bytes, size or Content-Type do not match the requested file.");
await taskAccess(principal, upload.taskId, principal.grant.purpose !== "agent", tx as unknown as Db, row.workSessionId ?? undefined);
const stored = upload.attachmentId ? null : await storage.putFile({ companyId: upload.companyId, namespace: "mcp-transfers", objectKey: upload.objectKey, originalFilename: upload.originalFilename, contentType: upload.contentType, body: bytes });
@@ -200,7 +210,11 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
objectKey: `${p.grant.companyId}/mcp-transfers/${id}`, storageProvider: storage.provider, expiresAt: new Date(Date.now() + lifetime) }).returning();
}
if (!found || found.argumentsHash !== hash) throw new McpCapabilityError("requestId was already used for a different upload.");
await tx.update(mcpAttachmentUploads).set({ expiresAt: new Date(Date.now() + lifetime), cleanedAt: null }).where(eq(mcpAttachmentUploads.id, found.id));
// A committed identity is immutable, including after user deletion.
// A rejected replay must not erase verified cleanup or permit new bytes.
if (!found.attachmentId) await tx.update(mcpAttachmentUploads)
.set({ expiresAt: new Date(Date.now() + lifetime), cleanedAt: null, cleanupVerifiedAt: null })
.where(eq(mcpAttachmentUploads.id, found.id));
return found;
});
if (upload.attachmentId) {
server/src/services/public-mcp/idle-history-proof.ts
+8 -5
View File
@@ -22,14 +22,17 @@ export const idleMcpHistoryChecks = [
OR (d.mailbox_item_id IS NOT NULL AND d.activity_id IS NULL AND EXISTS (
SELECT 1 FROM dot_mailbox_items m WHERE m.id = d.mailbox_item_id
AND m.company_id = s.company_id AND m.binding_id = s.binding_id))))) IS NOT TRUE` },
// The upload owner allocates object identity before storage writes and only
// stamps cleaned_at after deletion. A committed attachment is never deleted.
// Upload identity survives user deletion. Only the new owner-verified
// absence stamp can retire cleanup; legacy cleaned_at alone was optimistic.
{ reason: "mcp_attachment_uploads", query: `SELECT 1 FROM mcp_attachment_uploads u WHERE (
EXISTS (SELECT 1 FROM issues i WHERE i.id = u.task_id AND i.company_id = u.company_id)
EXISTS (SELECT 1 FROM companies c WHERE c.id = u.company_id)
AND NOT EXISTS (SELECT 1 FROM issues contradiction WHERE contradiction.id = u.task_id AND contradiction.company_id <> u.company_id)
AND (u.grant_id IS NULL OR EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = u.grant_id
AND g.company_id = u.company_id AND g.user_id = u.user_id))
AND ((u.attachment_id IS NULL AND u.cleaned_at IS NOT NULL)
OR (u.cleaned_at IS NULL AND u.attachment_id IS NOT NULL AND EXISTS (
AND ((u.cleaned_at IS NOT NULL AND u.cleanup_verified_at IS NOT NULL AND u.cleanup_verified_at >= u.cleaned_at
AND (u.attachment_id IS NULL OR NOT EXISTS (SELECT 1 FROM issue_attachments live WHERE live.id = u.attachment_id)))
OR (u.cleaned_at IS NULL AND u.cleanup_verified_at IS NULL AND u.attachment_id IS NOT NULL
AND EXISTS (SELECT 1 FROM issues i WHERE i.id = u.task_id AND i.company_id = u.company_id) AND EXISTS (
SELECT 1 FROM issue_attachments a JOIN assets stored ON stored.id = a.asset_id
WHERE a.id = u.attachment_id AND a.company_id = u.company_id AND a.issue_id = u.task_id
AND stored.company_id = u.company_id AND stored.object_key = u.object_key
server/src/services/public-mcp/upload-cleanup.ts
+38
View File
@@ -0,0 +1,38 @@
import { and, eq } from "drizzle-orm";
import { mcpAttachmentUploads, type Db } from "@paperclipai/db";
import type { StorageService } from "../../storage/types.js";
type AttachmentStorage = {
id: string;
companyId: string;
issueId: string;
provider: string;
objectKey: string;
};
export type McpAttachmentCleanup = AttachmentStorage & { verifiedAt: Date };
function uploadIdentity(attachment: AttachmentStorage) {
return and(eq(mcpAttachmentUploads.companyId, attachment.companyId),
eq(mcpAttachmentUploads.taskId, attachment.issueId), eq(mcpAttachmentUploads.attachmentId, attachment.id),
eq(mcpAttachmentUploads.storageProvider, attachment.provider), eq(mcpAttachmentUploads.objectKey, attachment.objectKey));
}
/** Called after deletion by the attachment owner. A resolved delete alone can
* be optimistic; only the exact upload's storage absence authorizes cleanup. */
export async function confirmMcpAttachmentCleanup(db: Db, storage: StorageService, attachment: AttachmentStorage): Promise<McpAttachmentCleanup | null> {
const [upload] = await db.select({ id: mcpAttachmentUploads.id }).from(mcpAttachmentUploads).where(uploadIdentity(attachment)).limit(1);
if (!upload) return null;
if (storage.provider !== attachment.provider || (await storage.headObject(attachment.companyId, attachment.objectKey)).exists !== false) {
throw new Error("MCP attachment storage cleanup could not be confirmed");
}
return { ...attachment, verifiedAt: new Date() };
}
/** Keep attachmentId as immutable replay identity. For attachment removal this
* runs in the metadata deletion transaction; issue removal stamps each object
* after its existing post-transaction storage cleanup completes. */
export async function recordMcpAttachmentCleanup(db: Pick<Db, "update">, cleanup: McpAttachmentCleanup | null | undefined) {
if (!cleanup) return;
await db.update(mcpAttachmentUploads).set({ cleanedAt: cleanup.verifiedAt, cleanupVerifiedAt: cleanup.verifiedAt }).where(uploadIdentity(cleanup));
}
server/src/services/voice/idle-history-proof.ts
+71
View File
@@ -0,0 +1,71 @@
import { VOICE_CALL_REPORT_WINDOW_MS } from "./voice-call-reports.js";
/** Available reports still own refresh work for the owner's whole 24h window.
* The accepted push receipt proves submission, never unobservable playback. */
export function completedVoiceAction(a: string): string {
const closedSession = (s: string) => `${s}.state IN ('ended', 'failed', 'expired') AND ${s}.ended_at IS NOT NULL
AND COALESCE(${s}.error_code, '') <> 'cleanup_pending'
AND (${s}.provider_session_id IS NULL OR ${s}.ended_at < now() - interval '${VOICE_CALL_REPORT_WINDOW_MS} milliseconds')`;
const submitted = (alias: string) => `(${alias}.status = 'completed' AND length(${alias}.result->>'messageId') > 0
AND ${alias}.result->>'playback' = 'unknown' AND ${alias}.result->>'acceptedAt' ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T')
OR (${alias}.status = 'call_ended' AND ${alias}.result->>'errorCode' = 'provider_unavailable')
OR (${alias}.status = 'failed' AND ${alias}.result->>'errorCode' IN
('credentials_rejected', 'insufficient_credits', 'provider_unavailable', 'voice_authority_unavailable'))`;
// Nonnull company/endpoint keys let PostgreSQL hash each debt set once,
// rather than rescanning every old session for every retained voice marker.
const settled = `(${a}.company_id, ${a}.endpoint_id) NOT IN (
SELECT session.company_id, session.endpoint_id FROM chat_voice_sessions session
WHERE (${closedSession('session')}) IS NOT TRUE)
AND (${a}.company_id, ${a}.endpoint_id) NOT IN (
SELECT admission.company_id, admission.endpoint_id FROM chat_voice_inbound_calls admission
WHERE admission.state NOT IN ('denied', 'expired', 'ended')
OR admission.cleanup_confirmed_at IS NULL OR admission.cleanup_next_attempt_at IS NOT NULL)
AND (${a}.company_id, ${a}.endpoint_id) NOT IN (
SELECT push.company_id, push.endpoint_id FROM chat_actions push
WHERE push.kind = 'speko_voice_reply_push' AND (${submitted('push')}) IS NOT TRUE)`;
// Configuration records this existing audit event only after tools, webhook,
// voice defaults and runtime setup all succeeded. A partial setup cannot use
// individual resource receipts to certify the rest of the owning operation.
const setupCommitted = `EXISTS (SELECT 1 FROM activity_log receipt JOIN chat_endpoints endpoint
ON endpoint.id = ${a}.endpoint_id AND endpoint.company_id = ${a}.company_id
WHERE receipt.company_id = ${a}.company_id AND receipt.entity_type = 'tool_connection'
AND receipt.entity_id = endpoint.connection_id::text AND receipt.details->>'endpointId' = endpoint.id::text
AND receipt.details->>'provider' = 'speko' AND receipt.action IN ('chat_endpoint.credentials_verified', 'chat_endpoint.reconnected')
AND receipt.created_at >= ${a}.updated_at)`;
return `EXISTS (SELECT 1 FROM chat_endpoints endpoint WHERE endpoint.id = ${a}.endpoint_id
AND endpoint.company_id = ${a}.company_id AND endpoint.provider = 'speko') AND ${settled} AND (
(${a}.kind = 'speko_tool_setup' AND ${a}.status = 'completed' AND ${setupCommitted}
AND ${a}.payload->>'name' IN ('submit_request', 'get_updates', 'answer_question')
AND ${a}.payload->>'fingerprint' ~ '^[a-f0-9]{64}$' AND ${a}.payload->>'ownershipDefinitionHash' ~ '^[a-f0-9]{64}$'
AND length(${a}.payload->>'ownershipAgentId') > 0 AND length(${a}.result->>'toolId') > 0
AND ${a}.provider_action_id = 'speko_tool:' || (${a}.payload->>'name') || ':' || (${a}.payload->>'fingerprint'))
OR (${a}.kind = 'speko_lifecycle_setup' AND ${a}.status = 'completed' AND ${setupCommitted}
AND ${a}.payload->>'fingerprint' ~ '^[a-f0-9]{64}$' AND length(${a}.payload->>'ownershipAgentId') > 0
AND ${a}.payload->>'ownershipUrl' LIKE 'https://%/events' AND length(${a}.result->>'webhookId') > 0
AND ${a}.provider_action_id = 'speko_lifecycle:' || (${a}.payload->>'fingerprint'))
OR (${a}.kind = 'speko_lifecycle_receipt' AND ${a}.status = 'completed' AND ${a}.result->>'accepted' = 'true'
AND ${a}.payload->>'type' IN ('call.status', 'call.report') AND ${a}.payload->>'fingerprint' ~ '^[a-f0-9]{64}$'
AND ${a}.provider_action_id LIKE 'speko_event:%' AND (
EXISTS (SELECT 1 FROM chat_voice_sessions session WHERE session.id::text = ${a}.payload->>'sessionId'
AND session.company_id = ${a}.company_id AND session.endpoint_id = ${a}.endpoint_id AND ${closedSession('session')})
OR (${a}.payload->'sessionId' = 'null'::jsonb AND EXISTS (
SELECT 1 FROM chat_voice_inbound_calls admission WHERE admission.company_id = ${a}.company_id
AND admission.endpoint_id = ${a}.endpoint_id AND admission.state IN ('denied', 'expired', 'ended')
AND admission.cleanup_confirmed_at IS NOT NULL AND admission.cleanup_next_attempt_at IS NULL))))
OR (${a}.kind = 'speko_voice_reply_push' AND (${submitted(a)})
AND ${a}.payload->>'textDigest' ~ '^[a-f0-9]{64}$'
AND ${a}.payload->>'partIndex' ~ '^[0-9]+$' AND ${a}.payload->>'partCount' ~ '^[1-9][0-9]*$'
AND CASE WHEN ${a}.payload->>'partIndex' ~ '^[0-9]+$' AND ${a}.payload->>'partCount' ~ '^[1-9][0-9]*$'
THEN (${a}.payload->>'partIndex')::numeric < (${a}.payload->>'partCount')::numeric ELSE false END
AND EXISTS (SELECT 1 FROM chat_voice_sessions session JOIN chat_voice_replies reply
ON reply.session_id = session.id AND reply.company_id = session.company_id
JOIN chat_publications publication ON publication.id = reply.publication_id AND publication.company_id = reply.company_id
WHERE session.id::text = ${a}.payload->>'sessionId' AND session.company_id = ${a}.company_id
AND session.endpoint_id = ${a}.endpoint_id AND session.conversation_id = ${a}.conversation_id
AND session.provider_session_id = ${a}.payload->>'providerSessionId' AND session.generation::text = ${a}.payload->>'generation'
AND reply.id::text = ${a}.payload->>'replyId' AND publication.id::text = ${a}.payload->>'publicationId'
AND ${a}.provider_action_id = 'voice_reply_push:' || reply.id::text || CASE WHEN ${a}.payload->>'partIndex' = '0'
THEN '' ELSE ':part:' || (${a}.payload->>'partIndex') END
AND publication.endpoint_id = session.endpoint_id AND publication.conversation_id = session.conversation_id
AND publication.issue_id = session.issue_id AND publication.state = 'published' AND publication.published_at IS NOT NULL AND ${closedSession('session')})))`;
}
server/src/services/voice/voice-call-reports.ts
+2 -1
View File
@@ -13,7 +13,8 @@ import { SpekoProviderError } from "./speko-provider.js";
type Entry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number };
type Report = SpekoCallReport;
const WINDOW_MS = 24 * 60 * 60 * 1000;
export const VOICE_CALL_REPORT_WINDOW_MS = 24 * 60 * 60 * 1000;
const WINDOW_MS = VOICE_CALL_REPORT_WINDOW_MS;
const eligible = (now: number) => and(inArray(chatVoiceSessions.state, ["ended", "failed", "expired"]),
isNotNull(chatVoiceSessions.providerSessionId), gte(chatVoiceSessions.endedAt, new Date(now - WINDOW_MS)));