fix(server): prove completed history before idle sleep (#18646)

Reviewed and verified at 592bce69c6.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-11-03 12:37:25 -06:00
1 parent 367f29478d
commit ccc63ea458
44 files changed
+4100 -112

No files matched your search

@@ -0,0 +1,49 @@
import { finishedIdleRun } from "./idle-work-proof.js";
/** Mirrors sweep/account: terminal provider status cannot settle usage. */
export function completedBrowserRun(run: string): string {
const hold = `${run}.result->'paperclipBudgetHold'`;
return `${run}.status IN ('completed', 'failed', 'cancelled')
AND ${run}.events_drained = 1 AND ${run}.recovery_cursor IS NULL
AND jsonb_typeof(${hold}) = 'object' AND ${hold}->>'invocationId' = ${run}.invocation_id::text
AND ((${run}.provider_run_id IS NOT NULL AND ${hold}->>'state' = 'settled'
AND ${hold}->>'accountedCentsExact' ~ '^[0-9]+([.][0-9]+)?$')
OR (${run}.provider_run_id IS NULL AND ${run}.status = 'failed' AND ${hold}->>'state' = 'released'))
-- account() replaces held authority with the authoritative terminal
-- summary and ledger charge. It retains prior diagnostic flags in JSON.
AND (${hold}->>'state' = 'settled' OR COALESCE(${hold}->>'billingUnknown', 'false') = 'false')
AND EXISTS (SELECT 1 FROM browser_use_sessions browser_parent_session JOIN heartbeat_runs browser_parent_run
ON browser_parent_run.id = ${run}.heartbeat_run_id AND browser_parent_run.company_id = browser_parent_session.company_id
AND browser_parent_run.agent_id = browser_parent_session.agent_id
WHERE browser_parent_session.id = ${run}.session_id AND browser_parent_session.company_id = ${run}.company_id
AND (browser_parent_run.native_issue_id = browser_parent_session.issue_id
OR browser_parent_run.context_snapshot->>'issueId' = browser_parent_session.issue_id::text
OR browser_parent_run.context_snapshot->>'taskId' = browser_parent_session.issue_id::text)
AND (browser_parent_run.native_issue_id IS NULL OR browser_parent_run.native_issue_id = browser_parent_session.issue_id)
AND (browser_parent_run.context_snapshot->>'issueId' IS NULL
OR browser_parent_run.context_snapshot->>'issueId' = browser_parent_session.issue_id::text)
AND (browser_parent_run.context_snapshot->>'taskId' IS NULL
OR browser_parent_run.context_snapshot->>'taskId' = browser_parent_session.issue_id::text)
AND ${finishedIdleRun('browser_parent_run')})`;
}
export const idleBrowserHistoryChecks = [
{ reason: "browser_use_sessions", query: `SELECT 1 FROM browser_use_sessions s WHERE (
(s.status = 'closed' OR (s.status = 'failed' AND s.provider_session_id IS NULL))
AND s.lease_token IS NULL AND s.lease_until IS NULL AND s.stop_requested IS NULL
AND EXISTS (SELECT 1 FROM issues i JOIN agents a ON a.id = s.agent_id AND a.company_id = i.company_id
JOIN tool_connections c ON c.id = s.connection_id AND c.company_id = i.company_id
JOIN connection_grants g ON g.id = s.grant_id AND g.company_id = i.company_id AND g.connection_id = c.id
WHERE i.id = s.issue_id AND i.company_id = s.company_id)
AND NOT EXISTS (SELECT 1 FROM browser_use_runs r WHERE r.session_id = s.id
AND (r.company_id <> s.company_id OR (${completedBrowserRun('r')}) IS NOT TRUE))
AND NOT EXISTS (SELECT 1 FROM browser_use_browsers b WHERE b.session_id = s.id
AND (b.company_id <> s.company_id OR b.status <> 'stopped'))) IS NOT TRUE` },
{ reason: "browser_use_runs", query: `SELECT 1 FROM browser_use_runs r WHERE (${completedBrowserRun('r')}) IS NOT TRUE` },
// stopped is written only after client.stop followed by client.browser
// confirms stopped. Parent/session/run checks remain independent barriers.
{ reason: "browser_use_browsers", query: `SELECT 1 FROM browser_use_browsers b WHERE (
b.status = 'stopped' AND EXISTS (SELECT 1 FROM browser_use_sessions s
WHERE s.id = b.session_id AND s.company_id = b.company_id AND s.status = 'closed'
AND s.stop_requested IS NULL AND s.lease_token IS NULL AND s.lease_until IS NULL)) IS NOT TRUE` },
] as const;
@@ -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))))`;
}
@@ -0,0 +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_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')})))))`;
}
@@ -0,0 +1,129 @@
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. */
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 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})
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
AND w.idempotency_key = 'interaction:' || (${a}.payload->>'interactionId') || ':' || (${a}.payload->>'interactionStatus')
AND ${finishedIdleWake('w')}))
WHEN ${a}.kind = 'failed_run_retry' THEN ${a}.status = 'processed'
AND ${a}.result->>'code' = 'failed_chat_retry_durable' AND ${wake}
WHEN ${a}.kind = 'provider_effect' THEN
(${a}.status = 'processed' AND length(${a}.result->>'providerMessageId') > 0 AND length(${a}.result->>'threadId') > 0)
OR (${a}.status = 'cancelled' AND ${a}.result->>'code' = 'provider_effect_no_longer_authorized')
OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'provider_effect_failed' AND ${a}.result->>'retryable' = 'false')
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 (${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))
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, completedTeamsAction, completedPhotonAction, completedDiscordAction, completedVoiceAction]
.map(proof => `(${proof(a)})`).join(' OR ')}) END`;
}
export const idleChatHistoryChecks = [
{ 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.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 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))
AND EXISTS (SELECT 1 FROM chat_conversations c JOIN chat_endpoints e
ON e.id = c.endpoint_id AND e.company_id = c.company_id
JOIN issues i ON i.id = p.issue_id AND i.company_id = c.company_id
WHERE c.id = p.conversation_id AND c.company_id = p.company_id AND c.endpoint_id = p.endpoint_id)
AND NOT EXISTS (SELECT 1 FROM chat_endpoint_leases l WHERE l.company_id = p.company_id
AND l.endpoint_id = p.endpoint_id AND l.lease_key LIKE 'publication:' || p.id::text || ':%' AND l.expires_at > now())
AND NOT EXISTS (SELECT 1 FROM chat_teams_file_transfers f WHERE f.publication_id = p.id
AND (f.company_id <> p.company_id OR f.phase NOT IN ('delivered', 'declined', 'expired', 'cancelled')
OR f.attempt_id IS NOT NULL OR f.attempt_expires_at IS NOT NULL))) IS NOT TRUE` },
{ reason: "chat_actions", query: `SELECT 1 FROM chat_actions a WHERE (${completedAction('a')}) IS NOT TRUE` },
{ reason: "chat_completion_deliveries", query: `SELECT 1 FROM chat_completion_deliveries d WHERE (
d.status IN ('delivered', 'superseded', 'exhausted')
AND EXISTS (SELECT 1 FROM chat_task_handoffs h JOIN issues task ON task.id = h.task_id AND task.company_id = h.company_id
JOIN issues source ON source.id = h.conversation_id AND source.company_id = h.company_id
WHERE h.task_id = d.task_id AND h.company_id = d.company_id
AND source.conversation_agent_id = h.agent_id
AND (d.status <> 'delivered' OR (d.status_version <= task.status_version
AND EXISTS (SELECT 1 FROM issue_comments c WHERE c.id = d.response_comment_id
AND c.company_id = d.company_id AND c.issue_id = h.conversation_id AND c.author_agent_id = h.agent_id
AND c.created_by_run_id = d.target_run_id))))
AND (d.target_run_id IS NULL AND d.status <> 'delivered' OR EXISTS (
SELECT 1 FROM heartbeat_runs r JOIN chat_task_handoffs h ON h.task_id = d.task_id
WHERE r.id = d.target_run_id AND r.company_id = d.company_id AND r.agent_id = h.agent_id
AND h.company_id = d.company_id AND r.context_snapshot->>'conversationSessionGeneration' = h.session_generation::text
AND (r.native_issue_id = h.conversation_id OR r.context_snapshot->>'issueId' = h.conversation_id::text)
AND ${finishedIdleRun('r')}))) IS NOT TRUE` },
] as const;
@@ -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),
);
@@ -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)))`;
}
@@ -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();
}
@@ -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)`;
}
@@ -0,0 +1,77 @@
/** 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')
AND (${a}.result->>'outcome' <> 'updated' OR ${a}.result->>'sessionStatus' <> 'processing'))
OR (${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('slack_session_no_published_state', 'slack_session_destination_changed'))
OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'slack_session_sync_rejected')))
OR (${a}.kind = 'slash_task_start' AND ${a}.status = 'processed'
AND length(${a}.result->>'providerMessageId') > 0 AND length(${a}.result->>'threadId') > 0)
OR (${a}.kind = 'slack_file_upload_receipt' AND ${a}.status = 'processed'
AND ${a}.result->>'code' = 'slack_file_upload_identity_confirmed')
OR (${a}.kind = 'slack_session_stop' AND ${a}.status IN ('processed', 'cancelled')
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 (${completedSlackConnectorAction(a)})`;
}
@@ -0,0 +1,20 @@
export function completedTelegramAction(a: string): string {
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')
OR (${a}.status = 'failed' AND ${a}.result->>'retryable' = 'false'
AND ${a}.result->>'code' = 'telegram_maintenance_failed')))
OR (${a}.kind = 'telegram_draft' AND (
(${a}.status = 'processed' AND ${a}.result->>'phase' = 'published' AND length(${a}.result->>'providerMessageId') > 0)
OR (${a}.status = 'cancelled' AND (${a}.result->>'phase' = 'stopped'
OR ${a}.result->>'code' = 'telegram_draft_presentation_stopped'))))`;
}
@@ -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)`;
}
@@ -0,0 +1,45 @@
import { finishedIdleRun, finishedIdleWake } from "./idle-work-proof.js";
/** The same positive container proof is used by ordinary and scheduled sleep.
* waiting/in_review is set by settleConversationTurn, but is not sufficient:
* a user comment may commit before the wake is enqueued. */
export function passiveIdleConversation(i: "i"): string {
return `${i}.conversation_agent_id IS NOT NULL AND length(${i}.conversation_user_id) > 0
AND ${i}.conversation_state = 'waiting' AND ${i}.status = 'in_review'
AND ${i}.assignee_agent_id = ${i}.conversation_agent_id AND ${i}.assignee_user_id IS NULL
AND (${i}.conversation_user_id = 'local-board' OR EXISTS (SELECT 1 FROM company_memberships membership
WHERE membership.company_id = ${i}.company_id AND membership.principal_type = 'user'
AND membership.principal_id = ${i}.conversation_user_id AND membership.status = 'active'))
AND ${i}.conversation_session_generation >= 0
AND ${i}.checkout_run_id IS NULL AND ${i}.execution_run_id IS NULL
AND ${i}.monitor_next_check_at IS NULL AND ${i}.monitor_wake_requested_at IS NULL
AND (${i}.execution_policy IS NULL OR (jsonb_typeof(${i}.execution_policy) = 'object'
AND COALESCE(${i}.execution_policy->'monitor', 'null'::jsonb) = 'null'::jsonb))
AND (${i}.execution_state IS NULL OR (jsonb_typeof(${i}.execution_state) = 'object'
AND ${i}.execution_state->>'status' = 'idle'
AND COALESCE(${i}.execution_state->'monitor', 'null'::jsonb) = 'null'::jsonb
AND COALESCE(${i}.execution_state->'currentParticipant', 'null'::jsonb) = 'null'::jsonb
AND COALESCE(${i}.execution_state->'currentStageId', 'null'::jsonb) = 'null'::jsonb
AND COALESCE(${i}.execution_state->'reviewRequest', 'null'::jsonb) = 'null'::jsonb))
AND EXISTS (SELECT 1 FROM agents a JOIN companies c ON c.id = a.company_id
WHERE a.id = ${i}.conversation_agent_id AND a.company_id = ${i}.company_id
AND c.status = 'active' AND a.status IN ('idle', 'active', 'paused', 'terminated'))
AND NOT EXISTS (SELECT 1 FROM heartbeat_runs r WHERE
(r.native_issue_id = ${i}.id OR r.context_snapshot->>'issueId' = ${i}.id::text OR r.context_snapshot->>'taskId' = ${i}.id::text)
AND (r.company_id <> ${i}.company_id OR r.agent_id <> ${i}.conversation_agent_id OR (${finishedIdleRun('r')}) IS NOT TRUE))
AND NOT EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE
(w.payload->>'issueId' = ${i}.id::text OR w.payload->>'taskId' = ${i}.id::text)
AND (w.company_id <> ${i}.company_id OR w.agent_id <> ${i}.conversation_agent_id
OR (${finishedIdleWake('w')}) IS NOT TRUE))
AND NOT EXISTS (SELECT 1 FROM issue_comments comment WHERE comment.issue_id = ${i}.id
AND comment.deleted_at IS NULL AND comment.author_user_id IS NOT NULL AND comment.author_agent_id IS NULL
AND comment.origin = 'comment' AND comment.client_request_id IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${i}.company_id
AND w.agent_id = ${i}.conversation_agent_id AND w.idempotency_key = 'conversation-comment:' || comment.id::text
AND ${finishedIdleWake('w')}))
AND NOT EXISTS (SELECT 1 FROM issue_thread_interactions interaction WHERE interaction.issue_id = ${i}.id
AND (interaction.company_id <> ${i}.company_id OR interaction.status = 'pending'))
AND NOT EXISTS (SELECT 1 FROM chat_completion_deliveries d JOIN chat_task_handoffs h ON h.task_id = d.task_id
WHERE h.conversation_id = ${i}.id AND (h.company_id <> ${i}.company_id OR d.company_id <> ${i}.company_id
OR d.status NOT IN ('delivered', 'superseded', 'exhausted')))`;
}
+134
View File
@@ -0,0 +1,134 @@
import { finishedIdleRun } from "./idle-work-proof.js";
/** executeDecisionEffects visits every selected ordinal, including failures.
* Its final aggregate cannot prove a missing row or another option's effect. */
function completedEffectManifest(d: string): string {
const options = `jsonb_array_elements(CASE WHEN jsonb_typeof(${d}.options) = 'array' THEN ${d}.options ELSE '[]'::jsonb END)`;
return `EXISTS (SELECT 1 FROM ${options} option_row WHERE option_row->>'id' = ${d}.chosen_option_id
AND jsonb_typeof(option_row->'effects') = 'array'
AND EXISTS (SELECT 1 FROM activity_log selection WHERE selection.company_id = ${d}.company_id
AND selection.entity_type = 'decision' AND selection.entity_id = ${d}.id::text
AND selection.action IN ('decision.decided', 'decision.dismissed')
AND selection.details->>'chosenOptionId' = ${d}.chosen_option_id
AND selection.details->>'decidedByUserId' = ${d}.decided_by_user_id)
AND (SELECT count(*) FROM ${options} duplicate_option WHERE duplicate_option->>'id' = ${d}.chosen_option_id) = 1
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(option_row->'effects') WITH ORDINALITY expected(effect, ordinal)
WHERE NOT EXISTS (SELECT 1 FROM decision_effect_executions execution JOIN issues target
ON target.id = execution.target_issue_id AND target.company_id = ${d}.company_id
JOIN activity_log audit ON audit.id = execution.activity_log_id AND audit.company_id = ${d}.company_id
AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text
AND audit.details->>'chosenOptionId' = ${d}.chosen_option_id
AND audit.details->>'executionId' = execution.id::text
WHERE execution.decision_id = ${d}.id AND execution.effect_index = expected.ordinal - 1
AND execution.effect_type = expected.effect->>'type' AND execution.target_issue_id::text = expected.effect->>'targetIssueId'
AND execution.status IN ('executed', 'failed', 'skipped') AND execution.executed_at IS NOT NULL
AND audit.action = 'decision.effect_' || execution.status))
AND NOT EXISTS (SELECT 1 FROM decision_effect_executions execution WHERE execution.decision_id = ${d}.id
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(option_row->'effects') WITH ORDINALITY expected(effect, ordinal)
WHERE execution.effect_index = expected.ordinal - 1 AND execution.effect_type = expected.effect->>'type'
AND execution.target_issue_id::text = expected.effect->>'targetIssueId'))
AND CASE WHEN ${d}.execution_status = 'succeeded' THEN NOT EXISTS (SELECT 1 FROM decision_effect_executions e
WHERE e.decision_id = ${d}.id AND e.status <> 'executed')
WHEN ${d}.execution_status = 'partial' THEN EXISTS (SELECT 1 FROM decision_effect_executions e
WHERE e.decision_id = ${d}.id AND e.status = 'executed') AND EXISTS (SELECT 1 FROM decision_effect_executions e
WHERE e.decision_id = ${d}.id AND e.status IN ('failed', 'skipped'))
ELSE NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id AND e.status = 'executed')
AND EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id AND e.status IN ('failed', 'skipped')) END)`;
}
function completedDecisionOutcome(d: string): string {
return `(${d}.status IN ('expired', 'cancelled') AND ${d}.execution_status IS NULL AND ${d}.chosen_option_id IS NULL
AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id)
AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id
AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.' || ${d}.status))
OR (${d}.status = 'decided' AND ${d}.decided_at IS NOT NULL AND ${d}.execution_status IN ('succeeded', 'partial', 'failed')
AND ((COALESCE(${d}.metadata->>'kind', '') <> 'attention_archive_proposal' AND (${completedEffectManifest(d)}))
OR (${completedArchiveOutcome(d)})
OR (${d}.chosen_option_id = 'dismissed' AND ${d}.execution_status = 'succeeded' AND ${d}.metadata->>'dismissed' = 'true'
AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id)
AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id
AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.dismissed'
AND audit.details->>'chosenOptionId' = 'dismissed' AND audit.details->>'decidedByUserId' = ${d}.decided_by_user_id))
OR (${d}.metadata->>'effectRecoveryError' IN ('missing_decider', 'missing_option')
AND ${d}.execution_status IN ('failed', 'partial')
AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id
AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.recovery_failed'
AND audit.details->>'reason' = ${d}.metadata->>'effectRecoveryError'
AND audit.details->>'executionStatus' = ${d}.execution_status
AND audit.details->>'chosenOptionId' IS NOT DISTINCT FROM ${d}.chosen_option_id))))`;
}
/** Archive applies the reviewed set in one transaction. Every catch is final
* (execution_status=failed); it neither replays the set nor leaves per-effect
* claims. Continuation and notification queues remain independent checks. */
function completedArchiveOutcome(d: string): string {
return `${d}.metadata->>'kind' = 'attention_archive_proposal'
AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id)
AND EXISTS (SELECT 1 FROM jsonb_array_elements(CASE WHEN jsonb_typeof(${d}.options) = 'array'
THEN ${d}.options ELSE '[]'::jsonb END) selected WHERE selected->>'id' = ${d}.chosen_option_id
AND selected->'effects' = '[]'::jsonb
AND (SELECT count(*) FROM jsonb_array_elements(${d}.options) duplicate WHERE duplicate->>'id' = ${d}.chosen_option_id) = 1)
AND ((${d}.execution_status = 'failed' AND ${d}.chosen_option_id = 'archive'
AND jsonb_typeof(${d}.metadata->'archiveProposalError') = 'string'
AND length(trim(${d}.metadata->>'archiveProposalError')) > 0)
OR (${d}.execution_status = 'succeeded' AND ${d}.metadata->>'archiveProposalError' IS NULL
AND (${d}.chosen_option_id <> 'archive' OR (
jsonb_typeof(${d}.target_snapshots) = 'object' AND ${d}.target_snapshots <> '{}'::jsonb
AND ${d}.metadata->>'manifestHash' ~ '^[a-f0-9]{64}$'
AND NOT EXISTS (SELECT 1 FROM jsonb_each(${d}.target_snapshots) snapshot WHERE (
snapshot.key = 'attention:' || (snapshot.value->'attentionArchive'->>'sourceKind') || ':' || (snapshot.value->'attentionArchive'->>'sourceId')
AND snapshot.value->'attentionArchive'->>'companyId' = ${d}.company_id::text
AND jsonb_typeof(snapshot.value->'attentionArchive'->'expectedVersion') = 'number'
AND snapshot.value->'attentionArchive'->>'expectedVersion' ~ '^[1-9][0-9]*$'
AND EXISTS (SELECT 1 FROM decision_retention retention WHERE retention.company_id = ${d}.company_id
AND retention.source_kind = snapshot.value->'attentionArchive'->>'sourceKind'
AND retention.source_id = snapshot.value->'attentionArchive'->>'sourceId'
AND retention.version::numeric > (snapshot.value->'attentionArchive'->>'expectedVersion')::numeric)
AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id
AND audit.actor_id = 'decision-archive-proposal' AND audit.action = 'decision_retention.archived'
AND audit.entity_type = 'attention_source' AND audit.entity_id = snapshot.value->'attentionArchive'->>'sourceId'
AND audit.details->>'sourceKind' = snapshot.value->'attentionArchive'->>'sourceKind'
AND audit.created_at >= ${d}.decided_at AND audit.responsible_user_id = ${d}.decided_by_user_id)) IS NOT TRUE)))))`;
}
function completedDecision(alias: string): string {
return `(${completedDecisionOutcome(alias)})
AND ${alias}.continuation_policy IN ('none', 'wake_origin_agent')
AND jsonb_typeof(${alias}.metadata) = 'object'
AND COALESCE(${alias}.metadata->>'continuationPending', 'false') = 'false'
AND (${alias}.continuation_policy = 'none' OR ${alias}.metadata->>'continuationDeliveredAt' IS NOT NULL)
AND EXISTS (SELECT 1 FROM heartbeat_runs r JOIN issues i ON i.id = ${alias}.origin_issue_id
AND i.company_id = r.company_id WHERE r.id = ${alias}.origin_run_id
AND r.company_id = ${alias}.company_id AND r.agent_id = ${alias}.origin_agent_id AND ${finishedIdleRun('r')})
AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${alias}.id
AND (e.status NOT IN ('executed', 'failed', 'skipped') OR e.executed_at IS NULL))`;
}
export const idleDecisionHistoryChecks = [
{ reason: "decisions", query: `SELECT 1 FROM decisions d WHERE (${completedDecision('d')}) IS NOT TRUE` },
{ reason: "decision_effect_executions", query: `SELECT 1 FROM decision_effect_executions e WHERE (
e.status IN ('executed', 'failed', 'skipped') AND e.executed_at IS NOT NULL
AND EXISTS (SELECT 1 FROM decisions d JOIN issues i ON i.id = e.target_issue_id AND i.company_id = d.company_id
WHERE d.id = e.decision_id AND (${completedDecision('d')}))) IS NOT TRUE` },
{ reason: "decision_archive_notification_outbox", query: `SELECT 1 FROM decision_archive_notification_outbox o WHERE (
o.status = 'delivered' AND o.delivered_at IS NOT NULL
AND o.archive_version > 0 AND EXISTS (SELECT 1 FROM decision_retention retention
WHERE retention.company_id = o.company_id AND retention.source_kind = o.source_kind
AND retention.source_id = o.source_id AND retention.archive_version >= o.archive_version)
AND EXISTS (SELECT 1 FROM agents a JOIN issues i ON i.company_id = a.company_id
WHERE a.id = o.origin_agent_id AND i.id = o.origin_issue_id AND i.company_id = o.company_id)) IS NOT TRUE` },
{ reason: "company_secret_proposals", query: `SELECT 1 FROM company_secret_proposals p WHERE (
p.kind IN ('secret', 'binding') AND p.status IN ('approved', 'rejected', 'withdrawn', 'expired')
AND p.resolved_at IS NOT NULL AND p.value_ciphertext IS NULL AND p.ciphertext_scrubbed_at IS NOT NULL
AND EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = p.origin_run_id
AND r.company_id = p.company_id AND r.agent_id = p.proposed_by_agent_id AND ${finishedIdleRun('r')})
AND (p.origin_issue_id IS NULL OR EXISTS (SELECT 1 FROM issues i
WHERE i.id = p.origin_issue_id AND i.company_id = p.company_id))
AND (p.status <> 'approved' OR (p.kind = 'secret' AND EXISTS (SELECT 1 FROM company_secrets s
WHERE s.id = p.created_secret_id AND s.company_id = p.company_id))
OR (p.kind = 'binding' AND p.applied_binding_config_path = p.config_path AND EXISTS (
SELECT 1 FROM agents a WHERE a.id = p.target_id AND a.company_id = p.company_id)))
AND (p.interaction_id IS NULL OR EXISTS (SELECT 1 FROM issue_thread_interactions i
WHERE i.id = p.interaction_id AND i.company_id = p.company_id AND i.issue_id = p.origin_issue_id
AND i.status IN ('accepted', 'rejected', 'cancelled', 'expired') AND i.resolved_at IS NOT NULL))) IS NOT TRUE` },
] as const;
@@ -22,7 +22,17 @@ export function registerEnvironmentCleanupWork(
onSettled: () => { requested++; },
run: async () => {
if (!input.canRun()) { requested++; next = null; return; }
if (Date.now() < earliest) { next = earliest; return; }
if (Date.now() < earliest) {
next = earliest;
if (queue === DELIVERY_QUEUES.historicalCleanup) {
// A settled metadata invalidation is not itself cleanup debt. A
// cheap owner read can discharge it without advancing the provider
// sweep cadence; later notifications keep their own generation.
const generation = requested;
if (await deadline() === null) { next = null; reconciled = generation; }
}
return;
}
const generation = ++requested;
earliest = Date.now() + minimumSpacing;
await run();
+5 -2
View File
@@ -5,6 +5,7 @@ import { AGENT_STATUSES, issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONIT
import type { ScheduledWake } from "./idle-scheduled-work.js";
import { onboardingFirstTaskAwaitingUserSql } from "./recovery/onboarding-first-task-wait.js";
import { completedAgentUnblockRequest, responseDrivenIssueWait } from "./idle-work-proof.js";
import { passiveIdleConversation } from "./conversation-idle-proof.js";
const processGeneration = randomUUID();
const limit = 10_000;
@@ -31,6 +32,7 @@ export async function readIdleIssueSchedule(tx: Pick<Db, "execute">, ownerId: st
untouched_onboarding: boolean; response_driven_wait: boolean;
unblock_request_finished: boolean;
response_wait_generation: unknown;
passive_conversation: boolean;
}>(sql`SELECT i.id, i.company_id, i.status, i.updated_at, i.assignee_agent_id, i.assignee_user_id,
i.checkout_run_id, i.execution_run_id, i.execution_policy, i.execution_state, i.unblock_descriptor,
i.conversation_agent_id, i.monitor_next_check_at, i.monitor_wake_requested_at, i.monitor_attempt_count,
@@ -39,6 +41,7 @@ export async function readIdleIssueSchedule(tx: Pick<Db, "execute">, ownerId: st
(i.assignee_agent_id IS NULL OR a.id IS NOT NULL) AS scoped,
(${sql.raw(responseDrivenIssueWait('i'))}) AS response_driven_wait,
(${sql.raw(completedAgentUnblockRequest('i'))}) AS unblock_request_finished,
(${sql.raw(passiveIdleConversation('i'))}) AS passive_conversation,
(SELECT jsonb_agg(jsonb_build_array(w.id, w.updated_at, w.kind, w.status, w.source_run_id,
w.effective_resolver_policy, w.continuation_policy, w.addressee_agent_id, w.addressee_user_id) ORDER BY w.id)
FROM issue_thread_interactions w WHERE w.issue_id = i.id AND w.status = 'pending') AS response_wait_generation,
@@ -62,7 +65,7 @@ export async function readIdleIssueSchedule(tx: Pick<Db, "execute">, ownerId: st
if (!statuses.has(row.status) || row.company_status !== "active" || row.scoped !== true
|| (row.assignee_agent_id && !agentStatuses.has(row.agent_status ?? ""))
|| row.unsafe_dependencies !== false || row.checkout_run_id || row.execution_run_id
|| row.conversation_agent_id || row.monitor_wake_requested_at
|| (row.conversation_agent_id && row.passive_conversation !== true) || row.monitor_wake_requested_at
|| (row.execution_state !== null && state?.status !== "idle")
|| (row.execution_policy !== null && !policy)) return { blocked: true };
if (row.status === "blocked" && row.dependency_ready) return { blocked: true };
@@ -82,7 +85,7 @@ export async function readIdleIssueSchedule(tx: Pick<Db, "execute">, ownerId: st
}
if (policy?.monitor != null) return { blocked: true };
const unblock = record(row.unblock_descriptor);
const passive = row.status === "backlog"
const passive = row.passive_conversation === true || row.status === "backlog"
|| (row.status === "todo" && !row.assignee_agent_id)
|| (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true)
|| (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board"
+20 -19
View File
@@ -10,6 +10,23 @@ import { verifiedResourceCleanup } from "./cleanup-verification-proof.js";
import { companyHoldMismatch } from "../modules/agent-lifecycle/index.js";
import { pendingBudgetRunCancellationPredicate } from "./heartbeat/budget-cancellation.js";
import { nativeResponsePresentationPendingQuery } from "./native-runtime/native-response-presentation-query.js";
import { passiveIdleConversation } from "./conversation-idle-proof.js";
import { idleChatHistoryChecks } from "./chat-channels/idle-history-proof.js";
import { idleToolActionHistoryChecks } from "./tool-action-idle-proof.js";
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/index.js";
export const completedHistoryChecks = [
...idleChatHistoryChecks, ...idleToolActionHistoryChecks, ...idleBrowserHistoryChecks,
...idleDecisionHistoryChecks, ...idleWorkspaceHistoryChecks, ...idleMcpHistoryChecks,
...idleExternalAgentHistoryChecks,
] as const;
const ordinaryIssues = `SELECT 1 FROM issues i WHERE i.status NOT IN ('done', 'cancelled')
AND (${passiveIdleConversation('i')}) IS NOT TRUE`;
export interface IdleSleepDrainStatus {
ownerId?: string;
@@ -58,7 +75,7 @@ const WORK_CHECKS = [
state NOT IN ('saved', 'unchanged', 'resolved', 'unavailable', 'conflict')
OR process_stopped_at IS NULL OR next_attempt_at IS NOT NULL
OR receipt ? 'baseline' OR receipt->>'cleanupPending' = 'true'`),
workCheck("sandbox_creation_operations", `SELECT 1 FROM sandbox_creation_operations WHERE state = 'pending'`),
workCheck("sandbox_creation_operations", `SELECT 1 FROM sandbox_creation_operations WHERE state IN ('pending', 'needs_reconciliation')`),
workCheck("budget_reservations", `SELECT 1 FROM budget_reservations WHERE state NOT IN ('settled', 'released')`),
workCheck("budget_policies", `SELECT 1 FROM budget_policies WHERE enforcement_version > enforcement_delivered_version`),
workCheck("agents.lifecycle", `SELECT 1 FROM agents WHERE lifecycle_state IN
@@ -78,7 +95,7 @@ const WORK_CHECKS = [
SELECT 1 FROM heartbeat_runs r WHERE r.id = w.run_id
AND r.company_id = w.company_id AND r.agent_id = w.agent_id
AND ${finishedRun("r")})))`),
workCheck("issues", `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`),
workCheck("issues", ordinaryIssues),
// A completed issue can still need its first watchdog review, including a
// retry after immediate evaluation failed before creating a review/run.
workCheck("issue_watchdogs", `SELECT 1 FROM issue_watchdogs WHERE status <> 'disabled'`),
@@ -263,23 +280,7 @@ const WORK_CHECKS = [
OR effect.delivery_state <> 'delivered' OR effect.delivered_at IS NULL
OR effect.next_attempt_at IS NOT NULL))
) IS NOT TRUE`),
// These less common work sources fail closed on any retained state. Their
// terminal-state exceptions can be added with tests for the owning service.
...[
"chat_deliveries", "chat_publications", "chat_actions",
"chat_completion_deliveries", "tool_action_requests", "tool_action_deliveries",
"workspace_runtime_services",
"execution_workspace_runtime_leases",
"company_transfer_runs", "decisions", "decision_effect_executions",
"decision_archive_notification_outbox", "browser_use_sessions", "browser_use_runs",
"browser_use_browsers",
"company_secret_proposals",
"mcp_oauth_grants", "mcp_mutation_receipts", "mcp_event_subscriptions",
"mcp_event_deliveries", "mcp_attachment_uploads", "dot_agent_bindings",
"dot_runner_assignments", "dot_runner_operations", "dot_mailbox_items",
"muse_agent_bindings", "muse_credentials", "muse_runner_assignments",
"muse_runner_operations", "muse_mailbox_items", "muse_input_deliveries", "external_agent_holds",
].map((table) => workCheck(table, `SELECT 1 FROM ${table}`)),
...completedHistoryChecks.map(check => workCheck(check.reason, check.query)),
] as const;
function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus, now: number): boolean {
+7 -1
View File
@@ -1,3 +1,4 @@
import { recordMcpAttachmentCleanup, type McpAttachmentCleanup } from "./public-mcp/upload-cleanup.js";
import type { IssueDisplayStatus } from "@paperclipai/shared";
import { issueDisplayStatusSql } from "./issue-display-status.js";
import { notifyStrandedAgentStatus, notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
@@ -13953,7 +13954,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({
@@ -13983,6 +13984,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;
}),
@@ -10314,6 +10314,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; });
@@ -1,10 +1,11 @@
import { createHash, randomBytes, randomUUID } from "node:crypto";
import { pipeline } from "node:stream/promises";
import express, { Router } from "express";
import { and, eq, gt, isNull, lt, sql } from "drizzle-orm";
import { and, asc, eq, gt, isNull, lt, sql } from "drizzle-orm";
import { type Db, mcpAttachmentUploads, mcpFileTickets, mcpOauthGrants, issues } from "@paperclipai/db";
import { z } from "zod";
import { MAX_ATTACHMENT_BYTES, isAllowedContentType } from "../../attachment-types.js";
import { logger } from "../../middleware/logger.js";
import type { StorageService } from "../../storage/types.js";
import { authorizationService } from "../authorization.js";
import { issueService } from "../issues.js";
@@ -78,16 +79,42 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
await taskAccess(p, attachment.issueId, false, queryDb, sessionId);
return attachment;
}
async function sweep() {
// Each transfer service lives for the app lifetime. Advance this bounded
// traversal only after commit, and wrap so uncertain rows are retried.
let cleanupAfterId: string | undefined;
let pendingSweep: Promise<void> | undefined;
function sweep() {
if (!pendingSweep) pendingSweep = runSweep().finally(() => { pendingSweep = undefined; });
return pendingSweep;
}
async function runSweep() {
await db.delete(mcpFileTickets).where(lt(mcpFileTickets.expiresAt, new Date()));
// Serialize cleanup against completion/reissue. Never remove a committed 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 });
// Serialize cleanup against completion/reissue. Requalify old optimistic
// cleanup and orphaned removed attachments, never a still-linked file.
cleanupAfterId = await db.transaction(async tx => {
const abandoned = await tx.select().from(mcpAttachmentUploads).where(and(
eq(mcpAttachmentUploads.storageProvider, storage.provider),
isNull(mcpAttachmentUploads.cleanupVerifiedAt), lt(mcpAttachmentUploads.expiresAt, new Date()),
cleanupAfterId ? gt(mcpAttachmentUploads.id, cleanupAfterId) : undefined,
sql`(${mcpAttachmentUploads.attachmentId} IS NULL OR NOT EXISTS (
SELECT 1 FROM issue_attachments attachment WHERE attachment.id = ${mcpAttachmentUploads.attachmentId}))`,
)).orderBy(asc(mcpAttachmentUploads.id)).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));
try {
await storage.deleteObject(upload.companyId, upload.objectKey);
if ((await storage.headObject(upload.companyId, upload.objectKey)).exists !== false) {
throw new Error("MCP upload storage cleanup could not be confirmed");
}
} catch {
// Cleanup is opportunistic across companies. Retain this uncertainty
// without denying an unrelated caller's authorized file transfer.
logger.warn({ companyId: upload.companyId, uploadId: upload.id }, "MCP upload cleanup remains unverified");
continue;
}
const verifiedAt = new Date();
await tx.update(mcpAttachmentUploads).set({ cleanedAt: verifiedAt, cleanupVerifiedAt: verifiedAt }).where(eq(mcpAttachmentUploads.id, upload.id));
}
return abandoned.length === 20 ? abandoned.at(-1)!.id : undefined;
});
}
async function ticket(p: McpPrincipal, target: { uploadId?: string; attachmentId?: string }, sessionId?: string) {
@@ -133,6 +160,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 +228,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) {
@@ -214,6 +246,26 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
...(publication ? { publication } : {}) };
// An older upload can lack publication proof. Verify the same bytes
// through another bounded PUT; never manufacture a receipt from metadata.
await db.transaction(async tx => {
const [current] = await tx.select().from(mcpAttachmentUploads)
.where(eq(mcpAttachmentUploads.id, upload.id)).for("update");
if (!current || current.companyId !== p.grant.companyId || current.userId !== p.grant.userId
|| current.argumentsHash !== hash || current.workSessionId !== sessionId
|| current.grantId && current.grantId !== p.grant.id
|| current.attachmentId !== attachment.id || current.storageProvider !== storage.provider
|| current.cleanedAt || current.cleanupVerifiedAt) throw unavailable();
const owner = await workSession(p, sessionId, current.taskId, true, tx as unknown as Db);
const live = await attachmentAccess(p, attachment.id, tx as unknown as Db, sessionId);
if (!owner || live.issueId !== current.taskId || live.provider !== current.storageProvider
|| live.objectKey !== current.objectKey || live.originalFilename !== current.originalFilename
|| live.byteSize !== current.byteSize || live.sha256 !== current.sha256
|| live.originatingRunId !== owner.runId || live.createdByAgentId !== owner.agentId) throw unavailable();
// Keep the committed attachment and cleanup identity immutable. A
// concurrent deletion still removes that attachment/stamps this row;
// PUT rechecks both and never stores bytes for a committed upload.
await tx.update(mcpAttachmentUploads).set({ expiresAt: new Date(Date.now() + lifetime) })
.where(eq(mcpAttachmentUploads.id, current.id));
});
}
return { ...await ticket(p, { uploadId: upload.id }, sessionId), method: "PUT", headers: { "Content-Type": a.contentType }, uploadId: upload.id, maxBytes: MAX_ATTACHMENT_BYTES,
instructions: "Send the exact file bytes with the given Content-Type. Success saves the attachment and returns its ID. An agent upload also returns a verified publication receipt; cite its evidenceRef in your completion report. There is no separate upload completion tool. Do not share this temporary file credential. Reuse requestId after a lost response." };
@@ -0,0 +1,40 @@
/** Completion receipts are retained for idempotency. Fixed aliases only. */
export const idleMcpHistoryChecks = [
{ reason: "mcp_oauth_grants", query: `SELECT 1 FROM mcp_oauth_grants g WHERE
(g.revoked_at IS NOT NULL AND EXISTS (SELECT 1 FROM companies c WHERE c.id = g.company_id)) IS NOT TRUE` },
{ reason: "mcp_mutation_receipts", query: `SELECT 1 FROM mcp_mutation_receipts m WHERE (
m.status = 'completed' AND jsonb_typeof(m.result) = 'object'
AND EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = m.grant_id
AND g.company_id = m.company_id AND g.user_id = m.user_id)) IS NOT TRUE` },
// Stopped subscriptions no longer scan for events. Expiry alone is not a
// handoff of responsibility for active subscriptions or unfinished delivery.
{ reason: "mcp_event_subscriptions", query: `SELECT 1 FROM mcp_event_subscriptions s WHERE (
s.stopped_at IS NOT NULL AND EXISTS (SELECT 1 FROM mcp_oauth_grants g
WHERE g.id = s.grant_id AND g.company_id = s.company_id)) IS NOT TRUE` },
{ reason: "mcp_event_deliveries", query: `SELECT 1 FROM mcp_event_deliveries d WHERE (
d.finished_at IS NOT NULL AND (d.outcome IN ('delivered', 'filtered', 'inactive', 'delivery_exhausted')
OR (d.outcome ~ '^http_[1-5][0-9][0-9]$' AND d.outcome !~ '^http_2'
AND (d.attempts >= 6 OR (d.outcome NOT IN ('http_408', 'http_429') AND d.outcome !~ '^http_5'))))
AND EXISTS (SELECT 1 FROM mcp_event_subscriptions s WHERE s.id = d.subscription_id
AND EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = s.grant_id AND g.company_id = s.company_id)
AND ((d.activity_id IS NOT NULL AND d.mailbox_item_id IS NULL AND EXISTS (
SELECT 1 FROM activity_log a WHERE a.id = d.activity_id AND a.company_id = s.company_id))
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` },
// 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 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.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
AND stored.provider = u.storage_provider)))) IS NOT TRUE` },
] as const;
@@ -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));
}
@@ -13,13 +13,15 @@ import { createStorageService } from "../storage/service.js";
import { createS3StorageProvider } from "../storage/s3-provider.js";
import type { StorageService } from "../storage/types.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { idleWorkSnapshot } from "./task-admission.js";
import { issueRecoveryActionService } from "./issue-recovery-actions.js";
import { notifyRemoteWorkspaceSnapshots, registerRemoteWorkspaceSnapshotRetentionWork, remoteWorkspaceSnapshotRetentionService } from "./remote-workspace-snapshot-retention.js";
const url = process.env.PAPERCLIP_TEST_DATABASE_URL;
if (url && !/^remote_workspace_[a-z_]*test$/.test(new URL(url).pathname.slice(1))) throw new Error("Disposable remote_workspace_*test database required");
const historyUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL");
const url = historyUrl ?? process.env.PAPERCLIP_TEST_DATABASE_URL;
if (!historyUrl && url && !/^remote_workspace_[a-z_]*test$/.test(new URL(url).pathname.slice(1))) throw new Error("Disposable remote_workspace_*test database required");
const support = url ? { supported: true } : await getEmbeddedPostgresTestSupport();
async function versionedS3Fixture() {
@@ -95,7 +97,8 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () =>
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
const now = new Date("2026-10-30T12:00:00Z");
beforeAll(async () => {
if (url) { await applyPendingMigrations(url); db = createDb(url); }
if (historyUrl) db = createDb(historyUrl, { maxConnections: 4 });
else if (url) { await applyPendingMigrations(url); db = createDb(url); }
else { temporary = await startEmbeddedPostgresTestDatabase("workspace-snapshot-retention-"); db = createDb(temporary.connectionString); }
}, 90_000);
afterAll(async () => { if (url) await closeRegisteredClients(url, { scope: "database" }); else await temporary?.cleanup(); });
@@ -320,6 +323,13 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () =>
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog",
ownerType: "board", cause: "workspace_recovery", fingerprint: randomUUID(), nextAction: "Recover saved files" });
const owner = service(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
const remove = storage.deleteObject.bind(storage);
let secondDeleteHeld = false, releaseSecondDelete!: () => void;
const secondDeleteReleased = new Promise<void>(resolve => { releaseSecondDelete = resolve; });
storage.deleteObject = async (...args) => {
if (args[1] === f.keys[1]) { secondDeleteHeld = true; await secondDeleteReleased; }
await remove(...args);
};
const worker = registerRemoteWorkspaceSnapshotRetentionWork(coordinator, owner, () => {});
try {
await worker.ready;
@@ -328,8 +338,23 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () =>
const resolved = await issueRecoveryActionService(db).resolveActiveForIssue({ companyId: f.companyId, sourceIssueId: f.issueId,
status: "resolved", outcome: "restored", resolutionNote: "Verified saved files" });
expect(resolved?.status).toBe("resolved");
await vi.waitFor(async () => expect((await storage.headObject(f.companyId, f.keys[0]!)).exists).toBe(false));
expect(await owner.nextDeadlineAt()).toBeNull();
} finally { await worker.stop(); await coordinator.stop(); }
try {
await vi.waitFor(() => expect(secondDeleteHeld).toBe(true));
expect((await storage.headObject(f.companyId, f.keys[0]!)).exists).toBe(false);
expect((await storage.headObject(f.companyId, f.keys[1]!)).exists).toBe(true);
expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id)))[0])
.toMatchObject({ id: f.row.id, status: "delete_requested", deleteAfter: now });
expect(await owner.nextDeadlineAt()).toBe(now.getTime());
expect(idleWorkSnapshot().active).toBeGreaterThan(0);
} finally { releaseSecondDelete(); }
await vi.waitFor(async () => {
for (const key of f.keys) expect((await storage.headObject(f.companyId, key)).exists).toBe(false);
expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id)))[0])
.toMatchObject({ id: f.row.id, status: "deleted", deleteAfter: null });
expect(await owner.nextDeadlineAt()).toBeNull();
expect(idleWorkSnapshot().active).toBe(0);
expect(coordinator.nextWakeAt()).toBeNull();
});
} finally { releaseSecondDelete(); await worker.stop(); await coordinator.stop(); }
});
});
@@ -0,0 +1,28 @@
import { finishedIdleRun } from "./idle-work-proof.js";
function completedRequest(a: string): string {
return `${a}.status IN ('executed', 'failed', 'rejected', 'expired', 'cancelled') AND ${a}.resolved_at IS NOT NULL
AND EXISTS (SELECT 1 FROM tool_invocations v WHERE v.id = ${a}.invocation_id AND v.company_id = ${a}.company_id
AND v.issue_id IS NOT DISTINCT FROM ${a}.issue_id AND v.completed_at IS NOT NULL
AND v.status IN ('succeeded', 'failed', 'denied', 'cancelled', 'timed_out', 'rate_limited')
AND (v.run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = v.run_id
AND r.company_id = v.company_id AND ${finishedIdleRun('r')})))
AND (${a}.interaction_id IS NULL OR EXISTS (SELECT 1 FROM issue_thread_interactions i WHERE i.id = ${a}.interaction_id
AND i.company_id = ${a}.company_id AND i.issue_id = ${a}.issue_id AND i.resolved_at IS NOT NULL
AND i.payload->'toolAction'->>'actionRequestId' = ${a}.id::text
AND i.payload->'toolAction'->>'invocationId' = ${a}.invocation_id::text
AND i.status IN ('accepted', 'rejected', 'expired', 'cancelled')
AND (${a}.status = 'rejected' OR i.result->'toolAction'->>'status' = CASE
WHEN ${a}.status = 'cancelled' THEN 'expired' ELSE ${a}.status END)))`;
}
export const idleToolActionHistoryChecks = [
{ reason: "tool_action_requests", query: `SELECT 1 FROM tool_action_requests a WHERE (${completedRequest('a')}) IS NOT TRUE` },
// delivered_at is the domain's acknowledgment. It also retires outcomes
// after issue completion/reassignment. Accepted continuations stay covered by
// the global wake/run checks, including coalesced run accounting.
{ reason: "tool_action_deliveries", query: `SELECT 1 FROM tool_action_deliveries d WHERE (
d.delivered_at IS NOT NULL AND EXISTS (SELECT 1 FROM tool_action_requests a
WHERE a.id = d.action_request_id AND a.company_id = d.company_id
AND a.issue_id = d.issue_id AND a.interaction_id = d.interaction_id AND (${completedRequest('a')}))) IS NOT TRUE` },
] as const;
@@ -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')})))`;
}
@@ -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)));
@@ -0,0 +1,52 @@
import { finishedIdleRun } from "./idle-work-proof.js";
export const idleWorkspaceHistoryChecks = [
// Failed imports are intentionally resumable. Completed/cancelled rows can
// still owe spool deletion, independently inspected by idle-local-work.
{ reason: "company_transfer_runs", query: `SELECT 1 FROM company_transfer_runs t WHERE (
t.status IN ('completed', 'cancelled') AND t.finished_at IS NOT NULL
AND t.direction IN ('import', 'export') AND jsonb_typeof(t.container_ref) = 'object'
AND jsonb_typeof(t.completed_parts) = 'array'
AND (t.status = 'cancelled' OR (t.error IS NULL AND t.manifest_sha256 ~ '^[a-f0-9]{64}$'
AND CASE WHEN jsonb_typeof(t.manifest->'chunks') = 'array' AND jsonb_typeof(t.manifest->'blobs') = 'array' THEN
t.chunk_count = jsonb_array_length(t.manifest->'chunks') AND t.blob_count = jsonb_array_length(t.manifest->'blobs')
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements((t.manifest->'chunks') || (t.manifest->'blobs')) part
WHERE jsonb_typeof(part->'name') IS DISTINCT FROM 'string' OR NOT t.completed_parts ? (part->>'name'))
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(t.completed_parts) done
WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements((t.manifest->'chunks') || (t.manifest->'blobs')) part WHERE part->>'name' = done))
WHEN jsonb_typeof(t.manifest->'parts') = 'array' THEN
t.chunk_count = jsonb_array_length(t.manifest->'parts') AND t.blob_count = 0
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(t.manifest->'parts') WITH ORDINALITY part(value, ordinal)
WHERE NOT t.completed_parts ? ('part-' || (ordinal - 1)::text))
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(t.completed_parts) done
WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements(t.manifest->'parts') WITH ORDINALITY part(value, ordinal)
WHERE done = 'part-' || (ordinal - 1)::text))
ELSE false END))
AND (t.company_id IS NOT NULL OR (t.direction = 'import' AND t.status = 'cancelled'))
AND (t.company_id IS NULL OR EXISTS (SELECT 1 FROM companies c WHERE c.id = t.company_id))) IS NOT TRUE` },
{ reason: "workspace_runtime_services", query: `SELECT 1 FROM workspace_runtime_services s WHERE (
s.status = 'stopped' AND s.stopped_at IS NOT NULL AND s.provider = 'local_process'
AND s.exposure_handle IS NULL
AND (s.exposure IS NULL OR (jsonb_typeof(s.exposure) = 'object'
AND s.exposure->>'state' = 'removed' AND s.url IS NULL
AND COALESCE(s.exposure->'publicUrl', 'null'::jsonb) = 'null'::jsonb))
AND (s.execution_workspace_id IS NULL OR EXISTS (SELECT 1 FROM execution_workspaces w
WHERE w.id = s.execution_workspace_id AND w.company_id = s.company_id))
AND (s.issue_id IS NULL OR EXISTS (SELECT 1 FROM issues i WHERE i.id = s.issue_id AND i.company_id = s.company_id))
AND (s.started_by_run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r
WHERE r.id = s.started_by_run_id AND r.company_id = s.company_id AND ${finishedIdleRun('r')}))) IS NOT TRUE` },
// This is an exclusivity row, not a provider lease. The domain only
// evaluates staleness on the next claim; no expiry job is owed by history.
{ reason: "execution_workspace_runtime_leases", query: `SELECT 1 FROM execution_workspace_runtime_leases l WHERE (
EXISTS (SELECT 1 FROM execution_workspaces w WHERE w.id = l.execution_workspace_id AND w.company_id = l.company_id)
AND ((l.owner_issue_id IS NOT NULL AND l.owner_key = 'issue:' || l.owner_issue_id::text
AND EXISTS (SELECT 1 FROM issues i WHERE i.id = l.owner_issue_id AND i.company_id = l.company_id
AND i.status IN ('done', 'cancelled')))
OR (l.owner_issue_id IS NULL AND l.owner_run_id IS NOT NULL AND l.owner_key = 'run:' || l.owner_run_id::text
AND EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = l.owner_run_id AND r.company_id = l.company_id
AND r.agent_id = l.owner_agent_id AND ${finishedIdleRun('r')})))
AND (l.owner_run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = l.owner_run_id
AND r.company_id = l.company_id AND r.agent_id = l.owner_agent_id AND ${finishedIdleRun('r')}))
AND NOT EXISTS (SELECT 1 FROM workspace_runtime_services s WHERE s.execution_workspace_id = l.execution_workspace_id
AND (s.company_id <> l.company_id OR s.status <> 'stopped' OR s.stopped_at IS NULL))) IS NOT TRUE` },
] as const;
+57 -2
View File
@@ -46,6 +46,7 @@ import {
import { and, desc, eq, gte, inArray, isNull, lte, ne, or, sql } from "drizzle-orm";
import { asNumber, asString, parseObject, renderTemplate } from "../adapters/utils.js";
import { conflict, unprocessable } from "../errors.js";
import { registerIdleLocalWorkProbe, type IdleLocalWork } from "./idle-local-work.js";
import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js";
import { resolveHomeAwarePath } from "../home-paths.js";
import { hasVerifiedWorktreeSeedManifest, isVerifiedWorktreeSeedManifest } from "../worktree-seed-manifest.js";
@@ -283,6 +284,23 @@ const quarantinedRuntimeExposurePorts = new Set<number>();
* companion together or not at all.
*/
const exposurePortPairClaims = new ExposurePortPairClaims();
let runtimeReconstructionComplete = false;
let runtimeStopsInFlight = 0;
let runtimeReconstructionEpoch = 0;
let runtimeReconciliation: ReturnType<typeof reconstructRuntimeServices> | null = null;
/** Read owner memory only. Durable stopped rows do not describe a surviving
* registry process, an allocation being published, or unfinished exposure cleanup. */
export function readWorkspaceRuntimeIdleWork(): IdleLocalWork {
if (!runtimeReconstructionComplete || runtimeReconciliation !== null) return "unknown";
return runtimeStopsInFlight || runtimeServicesById.size || runtimeProvisionByWorkspace.size || runtimeControlStartByOwner.size
|| runtimeReplacementClaimsByReuseKey.size || quarantinedRuntimeExposurePorts.size
|| exposurePortPairClaims.activePorts().size
|| [...inFlightAllocatedPorts.values()].some(until => until > Date.now()) ? "present" : "none";
}
// This registry and its probe share the application process lifetime, like the
// native-session owner. Stopping one resource must not unregister its siblings.
registerIdleLocalWorkProbe(readWorkspaceRuntimeIdleWork);
/**
* Execution-workspace statuses that still hold an exclusive lease. `archived`
* is the only terminal state; everything else — including `idle` — is a lane an
@@ -525,6 +543,8 @@ type ProcessOutputAccumulator = {
export async function resetRuntimeServicesForTests(
opts: { terminateProcesses?: boolean; simulateSupervisorExit?: boolean } = {},
) {
runtimeReconstructionComplete = false;
runtimeReconstructionEpoch += 1;
if (opts.terminateProcesses) {
for (const serviceId of [...runtimeServicesById.keys()]) {
await stopRuntimeService(serviceId).catch(() => undefined);
@@ -7155,7 +7175,21 @@ async function cleanupRecordExposure(
await persistRuntimeServiceRecord(record.db, record).catch(() => undefined);
}
async function withRuntimeStopWork<T>(stop: () => Promise<T>): Promise<T> {
runtimeStopsInFlight += 1;
try { return await stop(); }
catch (error) {
runtimeReconstructionComplete = false;
runtimeReconstructionEpoch += 1;
throw error;
} finally { runtimeStopsInFlight -= 1; }
}
async function stopRuntimeService(serviceId: string) {
return withRuntimeStopWork(() => stopRuntimeServiceRecord(serviceId));
}
async function stopRuntimeServiceRecord(serviceId: string) {
const record = runtimeServicesById.get(serviceId);
if (!record) return;
clearIdleTimer(record);
@@ -8296,7 +8330,11 @@ export async function releaseRuntimeServicesForRun(runId: string) {
}
}
export async function stopRuntimeServicesForExecutionWorkspace(input: {
export async function stopRuntimeServicesForExecutionWorkspace(input: Parameters<typeof stopRuntimeServicesForExecutionWorkspaceOwned>[0]) {
return withRuntimeStopWork(() => stopRuntimeServicesForExecutionWorkspaceOwned(input));
}
async function stopRuntimeServicesForExecutionWorkspaceOwned(input: {
db?: Db;
executionWorkspaceId: string;
workspaceCwd?: string | null;
@@ -8351,7 +8389,11 @@ export async function stopRuntimeServicesForExecutionWorkspace(input: {
}
}
export async function stopRuntimeServicesForProjectWorkspace(input: {
export async function stopRuntimeServicesForProjectWorkspace(input: Parameters<typeof stopRuntimeServicesForProjectWorkspaceOwned>[0]) {
return withRuntimeStopWork(() => stopRuntimeServicesForProjectWorkspaceOwned(input));
}
async function stopRuntimeServicesForProjectWorkspaceOwned(input: {
db?: Db;
projectWorkspaceId: string;
runtimeServiceId?: string | null;
@@ -8577,6 +8619,19 @@ export async function refreshPersistedRuntimeServiceHealth(input: {
}
export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
if (runtimeReconciliation) return runtimeReconciliation;
runtimeReconstructionComplete = false;
const epoch = ++runtimeReconstructionEpoch;
runtimeReconciliation = reconstructRuntimeServices(db).then(result => {
if (epoch === runtimeReconstructionEpoch) {
runtimeReconstructionComplete = result.restartFailed === 0 && result.exposureReservationDrift.length === 0;
}
return result;
}).finally(() => { runtimeReconciliation = null; });
return runtimeReconciliation;
}
async function reconstructRuntimeServices(db: Db) {
const rows = await db
.select()
.from(workspaceRuntimeServices)