Merge remote-tracking branch 'origin/staging' into fix/chat-code-not-copied-inline

Author: j15z

apps/desktop/package.json

@@ -9,7 +9,7 @@
   "type": "module",
   "main": "dist/main.cjs",
   "engines": {
-    "bun": ">=1.2.13",
+    "bun": ">=1.3.14",
     "node": ">=20.0.0"
   },
   "scripts": {

apps/docs/content/docs/en/cli/commands.mdx

@@ -113,3 +113,29 @@ sim configure [options]
 | `--unset <key...>` | No | Remove settings (endpoint, workspace, output). |
 
 </CommandTable>
+
+## Ask Sim and print the reply
+
+```bash
+sim chat <message> [options]
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `message` | Yes | What to ask Sim |
+
+</CommandTable>
+
+**Options**
+
+<CommandTable>
+
+| Option | Required | Description |
+| --- | --- | --- |
+| `-c, --conversation <id>` | No | Continue the conversation with this ID. |
+
+</CommandTable>

apps/docs/content/docs/en/cli/reference.mdx

@@ -101,6 +101,34 @@ sim configure [options]
 
 </CommandTable>
 
+## sim chat
+
+Ask Sim and print the reply
+
+```bash
+sim chat <message> [options]
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `message` | Yes | What to ask Sim |
+
+</CommandTable>
+
+**Options**
+
+<CommandTable>
+
+| Option | Required | Description |
+| --- | --- | --- |
+| `-c, --conversation <id>` | No | Continue the conversation with this ID. |
+
+</CommandTable>
+
 ## sim profiles
 
 Also spelled `sim profile`.

apps/realtime/package.json

@@ -5,7 +5,7 @@
   "license": "Apache-2.0",
   "type": "module",
   "engines": {
-    "bun": ">=1.2.13",
+    "bun": ">=1.3.14",
     "node": ">=20.0.0"
   },
   "scripts": {

apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts

@@ -396,8 +396,9 @@ describe('stale execution cleanup deadline grace', () => {
     const response = await GET(createRequest())
 
     expect(response.status).toBe(200)
-    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(8)
-    expect(dbChainMockFns.for).toHaveBeenCalledTimes(8)
+    // Nine batched arms: the connector sync-log retention pass is the newest.
+    expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(9)
+    expect(dbChainMockFns.for).toHaveBeenCalledTimes(9)
     for (const [strength, options] of dbChainMockFns.for.mock.calls) {
       expect(strength).toBe('update')
       expect(options).toEqual({ skipLocked: true })
@@ -469,7 +470,7 @@ describe('stale execution cleanup deadline grace', () => {
     const limits = dbChainMockFns.limit.mock.calls.map(([limit]) => limit)
     expect(limits.filter((limit) => limit === 100)).toHaveLength(20)
     expect(limits.filter((limit) => limit === 1000)).toHaveLength(30)
-    expect(limits.filter((limit) => limit === 2000)).toHaveLength(11)
+    expect(limits.filter((limit) => limit === 2000)).toHaveLength(12)
 
     const workflowUpdates = dbChainMockFns.update.mock.calls.filter(
       ([table]) => table === workflowExecutionLogs

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -1,6 +1,7 @@
 import { db } from '@sim/db'
 import {
   asyncJobs,
+  knowledgeConnectorSyncLog,
   tableJobs,
   workflowDeploymentOperation,
   workflowExecutionLogs,
@@ -31,6 +32,7 @@ import {
   STALE_SWEEPABLE_EXECUTION_STATUSES,
   type StaleSweepableExecutionStatus,
 } from '@/lib/logs/types'
+import { cancelStaleDispatches } from '@/lib/table/dispatcher'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import {
   carrierNotIrrecoverableSql,
@@ -52,12 +54,33 @@ const EXECUTION_DEADLINE_ERROR = getTimeoutErrorMessage(undefined)
 const TABLE_JOB_STALE_THRESHOLD_MINUTES = 95
 /** Terminal table-jobs older than this are pruned; only the latest job per table is ever read. */
 const TABLE_JOB_RETENTION_HOURS = 24
+/**
+ * A table run dispatch whose holder has not made progress for this long is
+ * treated as dead. Same shape and window as the table-job threshold above: the
+ * 90-minute Trigger.dev task ceiling (`maxDuration` in `trigger.config.ts`) plus
+ * five minutes of cleanup grace, measured from the dispatcher's own per-window
+ * heartbeat rather than from when the run was requested.
+ */
+const TABLE_DISPATCH_STALE_THRESHOLD_MINUTES = 95
+/** Per-run ceiling on reaped dispatches, so one tick cannot fan out unbounded SSE. */
+const TABLE_DISPATCH_MAX_PER_RUN = 200
 /**
  * Terminal deployment operations older than this are pruned. Every reader of
  * this table is latest-generation-only, and idempotency keys only need to
  * survive a client retry window, so 30 days is generous.
  */
 const DEPLOYMENT_OPERATION_RETENTION_DAYS = 30
+/**
+ * Terminal connector sync logs older than this are pruned. Nothing pruned them
+ * before, so the table grew by one row per sync run forever — a connector on a
+ * fifteen-minute interval writes about 35,000 rows a year on its own. That cost
+ * lands on `loadPreviousListingObservation`, which reads the newest `completed`
+ * row per connector through an index covering `connector_id` alone, so every
+ * retained row makes the sort behind the deletion-safety corroboration slower.
+ */
+const CONNECTOR_SYNC_LOG_RETENTION_DAYS = 30
+const CONNECTOR_SYNC_LOG_PRUNE_BATCH_SIZE = 2000
+const CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN = 20_000
 const DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE = 2000
 const DEPLOYMENT_OPERATION_PRUNE_MAX_BATCHES = 10
 const WORKFLOW_EXECUTION_MUTATION_BATCH_SIZE = 100
@@ -144,6 +167,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     const staleTableJobThreshold = new Date(
       now.getTime() - TABLE_JOB_STALE_THRESHOLD_MINUTES * 60 * 1000
     )
+    const staleDispatchThreshold = new Date(
+      now.getTime() - TABLE_DISPATCH_STALE_THRESHOLD_MINUTES * 60 * 1000
+    )
 
     let staleExecutionsFound = 0
     let cleaned = 0
@@ -538,6 +564,90 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    /**
+     * Prune terminal connector sync logs past retention.
+     *
+     * HARD INVARIANT: the newest row per connector must survive, and so must the
+     * newest `completed` row. `loadPreviousListingObservation` reconstructs the
+     * previous listing from the latest `completed` log, and that reconstruction
+     * decides whether a suspect listing is corroborated — i.e. whether
+     * reconciliation may delete documents. Pruning the last `completed` row
+     * would silently change deletion behaviour, so both `exists` guards below
+     * are load-bearing rather than defensive.
+     *
+     * `started` rows are never eligible: they are either in flight or waiting on
+     * the scheduler's own sweep to close them.
+     */
+    let connectorSyncLogsPruned = 0
+    try {
+      const syncLogRetention = new Date(
+        Date.now() - CONNECTOR_SYNC_LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000
+      )
+      const newerSyncLog = alias(knowledgeConnectorSyncLog, 'newer_sync_log')
+      const newerCompletedSyncLog = alias(knowledgeConnectorSyncLog, 'newer_completed_sync_log')
+      const syncLogPredicate = and(
+        inArray(knowledgeConnectorSyncLog.status, ['completed', 'failed']),
+        lt(knowledgeConnectorSyncLog.startedAt, syncLogRetention),
+        exists(
+          db
+            .select({ id: newerSyncLog.id })
+            .from(newerSyncLog)
+            .where(
+              and(
+                eq(newerSyncLog.connectorId, knowledgeConnectorSyncLog.connectorId),
+                gt(newerSyncLog.startedAt, knowledgeConnectorSyncLog.startedAt)
+              )
+            )
+        ),
+        or(
+          ne(knowledgeConnectorSyncLog.status, 'completed'),
+          exists(
+            db
+              .select({ id: newerCompletedSyncLog.id })
+              .from(newerCompletedSyncLog)
+              .where(
+                and(
+                  eq(newerCompletedSyncLog.connectorId, knowledgeConnectorSyncLog.connectorId),
+                  eq(newerCompletedSyncLog.status, 'completed'),
+                  gt(newerCompletedSyncLog.startedAt, knowledgeConnectorSyncLog.startedAt)
+                )
+              )
+          )
+        )
+      )
+      const syncLogResult = await runBatchedMutation({
+        batchSize: CONNECTOR_SYNC_LOG_PRUNE_BATCH_SIZE,
+        maxRowsPerRun: CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN,
+        claim: (tx, limit) =>
+          tx
+            .select({ id: knowledgeConnectorSyncLog.id })
+            .from(knowledgeConnectorSyncLog)
+            .where(syncLogPredicate)
+            .limit(limit)
+            .for('update', { skipLocked: true }),
+        mutation: (tx, candidateIds) =>
+          tx
+            .delete(knowledgeConnectorSyncLog)
+            .where(inArray(knowledgeConnectorSyncLog.id, candidateIds))
+            .returning({ id: knowledgeConnectorSyncLog.id }),
+      })
+      connectorSyncLogsPruned = syncLogResult.affected
+      if (connectorSyncLogsPruned > 0) {
+        logger.info(
+          `Pruned ${connectorSyncLogsPruned} old connector sync logs (retention: ${CONNECTOR_SYNC_LOG_RETENTION_DAYS}d)`
+        )
+      }
+      if (syncLogResult.reachedLimit) {
+        logger.info('Deferred remaining connector sync logs after reaching the per-run cap', {
+          maxRowsPerRun: CONNECTOR_SYNC_LOG_MAX_ROWS_PER_RUN,
+        })
+      }
+    } catch (error) {
+      logger.error('Failed to prune old connector sync logs:', {
+        error: toError(error).message,
+      })
+    }
+
     /**
      * Prune terminal deployment operations past retention. HARD INVARIANT:
      * the newest-generation row per workflow must always survive — the next
@@ -604,6 +714,29 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    /**
+     * Cancel table run dispatches abandoned by a dead dispatcher. Nothing else
+     * reclaims them — every other terminal transition is user- or flow-initiated
+     * — so a dispatcher killed mid-loop left the row `dispatching` forever and
+     * the client's "X running" overlay with it. Ages from the dispatcher's
+     * per-window heartbeat, so a slow-but-live dispatch is spared.
+     */
+    let staleDispatchesCancelled = 0
+    try {
+      staleDispatchesCancelled = (
+        await cancelStaleDispatches(staleDispatchThreshold, TABLE_DISPATCH_MAX_PER_RUN)
+      ).length
+      if (staleDispatchesCancelled > 0) {
+        logger.warn(`Cancelled ${staleDispatchesCancelled} abandoned table run dispatches`, {
+          thresholdMinutes: TABLE_DISPATCH_STALE_THRESHOLD_MINUTES,
+        })
+      }
+    } catch (error) {
+      logger.error('Failed to cancel abandoned table run dispatches:', {
+        error: toError(error).message,
+      })
+    }
+
     return NextResponse.json({
       success: true,
       executions: {
@@ -622,6 +755,14 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       tableJobs: {
         staleMarkedFailed: staleTableJobsMarkedFailed,
       },
+      connectorSyncLogs: {
+        pruned: connectorSyncLogsPruned,
+        retentionDays: CONNECTOR_SYNC_LOG_RETENTION_DAYS,
+      },
+      tableRunDispatches: {
+        staleCancelled: staleDispatchesCancelled,
+        thresholdMinutes: TABLE_DISPATCH_STALE_THRESHOLD_MINUTES,
+      },
       deploymentOperations: {
         pruned: deploymentOperationsPruned,
         retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS,

apps/sim/app/api/tools/cloudwatch/put-metric-data/retry-semantics.test.ts

@@ -0,0 +1,73 @@
+/**
+ * Pins what `maxAttempts` *means*, using the real AWS SDK.
+ *
+ * `route.test.ts` asserts the route configures `maxAttempts: 1`; on its own that
+ * only pins a number. This file counts how many datapoints a real
+ * `CloudWatchClient` actually hands to a peer that accepts the request and then
+ * dies before writing a response byte -- the ambiguous transport failure the
+ * SDK's retry layer replays, and the one that makes CloudWatch aggregate a
+ * duplicate. It fails if a future SDK bump changes the default budget or stops
+ * honouring the pin.
+ *
+ * Deliberately not mocking `@aws-sdk/client-cloudwatch` here: the SDK's retry
+ * middleware is the subject under test, so it must be the real one.
+ *
+ * @vitest-environment node
+ */
+import http from 'node:http'
+import type { AddressInfo } from 'node:net'
+import { CloudWatchClient, PutMetricDataCommand } from '@aws-sdk/client-cloudwatch'
+import { describe, expect, it } from 'vitest'
+
+/** `@smithy/util-retry`'s `DEFAULT_MAX_ATTEMPTS`, which every unpinned client inherits. */
+const SDK_DEFAULT_MAX_ATTEMPTS = 3
+
+async function countDeliveries(clientConfig: Record<string, unknown>): Promise<number> {
+  let received = 0
+  const server = http.createServer((req, res) => {
+    req.on('data', () => {})
+    req.on('end', () => {
+      if (String(req.headers['x-amz-target'] ?? '').endsWith('PutMetricData')) {
+        received++
+        req.socket.destroy()
+        return
+      }
+      res.writeHead(200, { 'content-type': 'application/x-amz-json-1.0' })
+      res.end('{}')
+    })
+  })
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
+  const { port } = server.address() as AddressInfo
+
+  const client = new CloudWatchClient({
+    region: 'us-east-1',
+    endpoint: `http://127.0.0.1:${port}`,
+    credentials: { accessKeyId: 'AKIAEXAMPLE', secretAccessKey: 'secret' },
+    ...clientConfig,
+  })
+
+  try {
+    await client.send(
+      new PutMetricDataCommand({
+        Namespace: 'Sim/Test',
+        MetricData: [{ MetricName: 'Requests', Value: 1 }],
+      })
+    )
+  } catch {
+    /* Every attempt fails by design; the delivery count is the assertion. */
+  } finally {
+    client.destroy()
+    await new Promise<void>((resolve) => server.close(() => resolve()))
+  }
+  return received
+}
+
+describe('aws sdk retry semantics for PutMetricData', () => {
+  it('delivers the datapoint exactly once when maxAttempts is pinned to 1', async () => {
+    await expect(countDeliveries({ maxAttempts: 1 })).resolves.toBe(1)
+  })
+
+  it('aggregates a duplicate for every retry the default budget allows', async () => {
+    await expect(countDeliveries({})).resolves.toBe(SDK_DEFAULT_MAX_ATTEMPTS)
+  })
+})