test(server): qualify completed history through domain writers

Bind retained transfers and chat completion receipts to exact owner evidence, qualify every retained-history family through normal producer paths, and fence runtime stop cleanup in memory.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/chat-channels/idle-history-proof.ts

@@ -44,6 +44,15 @@ function completedAction(a: string): string {
 export const idleChatHistoryChecks = [
   { reason: "chat_deliveries", query: `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.state <> 'processed' OR d.event_kind NOT IN ('mention', 'message', 'direct_message')
+      OR EXISTS (SELECT 1 FROM chat_actions consumed WHERE consumed.delivery_id = d.id
+        AND consumed.company_id = d.company_id AND consumed.endpoint_id = d.endpoint_id
+        AND consumed.kind = 'inbound_wakeup' AND (${completedAction('consumed')})))
     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))
@@ -68,12 +77,14 @@ export const idleChatHistoryChecks = [
       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 task.status = 'done'
+        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.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;

server/src/services/workspace-history-idle-proof.ts

@@ -7,6 +7,21 @@ export const idleWorkspaceHistoryChecks = [
     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 (

server/src/services/workspace-runtime.ts

@@ -285,6 +285,7 @@ const quarantinedRuntimeExposurePorts = new Set<number>();
  */
 const exposurePortPairClaims = new ExposurePortPairClaims();
 let runtimeReconstructionComplete = false;
+let runtimeStopsInFlight = 0;
 let runtimeReconstructionEpoch = 0;
 let runtimeReconciliation: ReturnType<typeof reconstructRuntimeServices> | null = null;
 
@@ -292,7 +293,7 @@ let runtimeReconciliation: ReturnType<typeof reconstructRuntimeServices> | null
  * registry process, an allocation being published, or unfinished exposure cleanup. */
 export function readWorkspaceRuntimeIdleWork(): IdleLocalWork {
   if (!runtimeReconstructionComplete || runtimeReconciliation !== null) return "unknown";
-  return runtimeServicesById.size || runtimeProvisionByWorkspace.size || runtimeControlStartByOwner.size
+  return runtimeStopsInFlight || runtimeServicesById.size || runtimeProvisionByWorkspace.size || runtimeControlStartByOwner.size
     || runtimeReplacementClaimsByReuseKey.size || quarantinedRuntimeExposurePorts.size
     || exposurePortPairClaims.activePorts().size
     || [...inFlightAllocatedPorts.values()].some(until => until > Date.now()) ? "present" : "none";
@@ -7174,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);
@@ -8315,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;
@@ -8370,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;