fix(v2): honour a requested stats window, and answer a claimed graph id with a conflict

Log statistics accepted a start and an end, filtered the totals by them, and
then built the series against wall-clock now. Bucket width was computed over a
span the caller never asked for, and every bucket past the requested end was
structurally empty - so a bounded historical query returned a wrong-width series
with fabricated trailing buckets, under a window label that disagreed with the
request. Each edge now honours the bound it was given and keeps its previous
derivation when omitted, so an unbounded request is unchanged.

Separately, block, edge and subflow ids are global primary keys while the
delete that precedes a state replace is scoped to one workflow. An id owned by
another workflow survived that delete, the insert violated the key, and because
callers pass their own transaction the driver error escaped unclassified as a
server fault. The write now refuses such an id up front with a conflict naming
it, and re-classifies the same violation if one races past the check, since the
lock covers only the workflow being written. The dry run checks the ids a commit
would insert and reports the warnings a commit would report, which is what its
own contract already promised.

Author: waleedlatif1

apps/sim/app/api/logs/stats/route.ts

@@ -72,7 +72,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       const whereCondition = commonFilters ? and(workspaceFilter, commonFilters) : workspaceFilter
 
       const bounds = await readLogStatsBounds(whereCondition)
-      const window = resolveLogStatsWindow(bounds, params.segmentCount)
+      const window = resolveLogStatsWindow(bounds, params.segmentCount, {
+        requestedStart: params.startDate ? new Date(params.startDate) : undefined,
+        requestedEnd: params.endDate ? new Date(params.endDate) : undefined,
+      })
       const rows = await readLogStatsSegments(
         whereCondition,
         window.startTime.toISOString(),

apps/sim/lib/api/contracts/v2/logs-stats.ts

@@ -105,7 +105,7 @@ export const v2LogStatsSchema = z
         end: v2TimestampSchema.describe('ISO 8601 end of the window.'),
       })
       .describe(
-        'The window the buckets span: the oldest matching run through the later of the newest matching run and now. A workspace with no matching runs reports the trailing 24 hours.'
+        'The window the buckets span. `startDate` and `endDate` are used verbatim when supplied; an omitted edge falls back to the oldest matching run on the left and to the later of the newest matching run and now on the right. With no matching runs the window is 24 hours wide, measured back from that right edge — the trailing 24 hours when no `endDate` was supplied, and the 24 hours preceding `endDate` when one was supplied without a `startDate`.'
       ),
     segmentMs: z.number().describe('Width of one bucket in milliseconds.'),
   })

apps/sim/lib/api/contracts/v2/openapi/logs.ts

@@ -207,7 +207,7 @@ const declaredRoutes = [
     logsOperation({
       operationId: 'getLogStats',
       summary: 'Get Log Statistics',
-      description: `Bucketed run counts, success rate, error count, and mean latency for a workspace and for each of its workflows — the aggregate a caller would otherwise have to page every run to compute. The window spans the oldest matching run through the later of the newest matching run and now, divided into exactly \`segmentCount\` equal buckets whose width is \`max(60000, floor(windowMs / segmentCount))\` milliseconds. The one-minute floor is a floor on bucket width, not on the window: when it applies, the series runs past \`timeBounds.end\` and the trailing buckets are empty rather than the window being compressed. A folder path covers its whole subtree. Per-workflow series are capped and \`workflowsTruncated\` reports whether the cap applied; the workspace totals are always computed from every workflow. ${RUN_RETENTION} ${FOLDER_TREE_TOO_LARGE}`,
+      description: `Bucketed run counts, success rate, error count, and mean latency for a workspace and for each of its workflows — the aggregate a caller would otherwise have to page every run to compute. The window spans \`startDate\` through \`endDate\` when both are supplied; an omitted edge falls back to the oldest matching run on the left and to the later of the newest matching run and now on the right. With no matching runs the window is 24 hours wide, measured back from that right edge — the trailing 24 hours when no \`endDate\` was supplied, and the 24 hours preceding \`endDate\` when one was supplied without a \`startDate\`. The window is divided into exactly \`segmentCount\` equal buckets whose width is \`max(60000, floor(windowMs / segmentCount))\` milliseconds. The one-minute floor is a floor on bucket width, not on the window: when it applies, the series runs past \`timeBounds.end\` and the trailing buckets are empty rather than the window being compressed. A folder path covers its whole subtree. Per-workflow series are capped and \`workflowsTruncated\` reports whether the cap applied; the workspace totals are always computed from every workflow. ${RUN_RETENTION} ${FOLDER_TREE_TOO_LARGE}`,
       errors: [...RESOURCE_ERRORS, 'PayloadTooLarge'],
       success: { description: 'Bucketed execution statistics for the workspace.' },
     }),

apps/sim/lib/logs/application/get-log-stats.ts

@@ -47,7 +47,10 @@ export const getLogStats = defineAuthorizedWorkspaceUseCase({
     )
 
     const bounds = await readLogStatsBounds(where)
-    const window = resolveLogStatsWindow(bounds, input.segmentCount)
+    const window = resolveLogStatsWindow(bounds, input.segmentCount, {
+      requestedStart: input.filters.startDate,
+      requestedEnd: input.filters.endDate,
+    })
     const rows = await readLogStatsSegments(where, window.startTime.toISOString(), window.segmentMs)
     return buildDashboardStats(rows, window, input.segmentCount, {
       maxWorkflows: MAX_STATS_WORKFLOWS,

apps/sim/lib/logs/application/log-analytics-use-cases.test.ts

@@ -129,6 +129,36 @@ describe('getLogStats', () => {
     )
   })
 
+  /**
+   * The wiring, not the arithmetic: `resolveLogStatsWindow` is exercised for
+   * real here, so a requested window that never reaches it shows up as both a
+   * wrong segment origin on the read and a wrong `timeBounds` on the response.
+   */
+  it('spans the requested window rather than the rows that happen to exist', async () => {
+    const { stats } = await getLogStats.execute({
+      principal: workspacePrincipal,
+      input: {
+        workspaceId: 'workspace-1',
+        filters: {
+          startDate: new Date('2026-08-01T00:00:00.000Z'),
+          endDate: new Date('2026-08-02T00:00:00.000Z'),
+        },
+        segmentCount: 2,
+      },
+    })
+
+    expect(mocks.readSegments).toHaveBeenCalledWith(
+      expect.anything(),
+      '2026-08-01T00:00:00.000Z',
+      12 * 60 * 60 * 1000
+    )
+    expect(stats.timeBounds).toEqual({
+      start: '2026-08-01T00:00:00.000Z',
+      end: '2026-08-02T00:00:00.000Z',
+    })
+    expect(stats.segmentMs).toBe(12 * 60 * 60 * 1000)
+  })
+
   it('resolves the folder scope only after authorization, and only when asked', async () => {
     await getLogStats.execute({
       principal: workspacePrincipal,

apps/sim/lib/logs/stats.test.ts

@@ -29,7 +29,7 @@ describe('resolveLogStatsWindow', () => {
   const now = new Date('2026-01-15T12:00:00.000Z')
 
   it('falls back to the trailing 24 hours when nothing ran', () => {
-    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 24, now)
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 24, { now })
 
     expect(resolved.endTime).toEqual(now)
     expect(resolved.startTime).toEqual(new Date('2026-01-14T12:00:00.000Z'))
@@ -39,7 +39,7 @@ describe('resolveLogStatsWindow', () => {
     const resolved = resolveLogStatsWindow(
       { minTime: '2026-01-15T00:00:00.000Z', maxTime: '2026-01-15T06:00:00.000Z' },
       12,
-      now
+      { now }
     )
 
     expect(resolved.endTime).toEqual(now)
@@ -50,17 +50,96 @@ describe('resolveLogStatsWindow', () => {
     const resolved = resolveLogStatsWindow(
       { minTime: '2026-01-15T12:00:00.000Z', maxTime: '2026-01-15T12:00:01.000Z' },
       500,
-      now
+      { now }
     )
 
     expect(resolved.segmentMs).toBe(60_000)
   })
 
   it('divides by segmentCount without producing a zero-width bucket', () => {
-    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 1, now)
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 1, { now })
 
     expect(resolved.segmentMs).toBe(24 * 60 * 60 * 1000)
   })
+
+  /**
+   * The `segmentMs` assertion is the load-bearing half: pinning `endTime`
+   * alone would still pass for a fix that relabelled `timeBounds` without
+   * re-deriving the bucket width the series is stamped from.
+   */
+  it('ends the window at the requested end rather than at now', () => {
+    const resolved = resolveLogStatsWindow(
+      { minTime: '2026-01-14T00:00:00.000Z', maxTime: '2026-01-14T06:00:00.000Z' },
+      12,
+      { requestedEnd: new Date('2026-01-14T12:00:00.000Z'), now }
+    )
+
+    expect(resolved.endTime).toEqual(new Date('2026-01-14T12:00:00.000Z'))
+    expect(resolved.segmentMs).toBe(60 * 60 * 1000)
+  })
+
+  it('starts the window at the requested start rather than at the oldest run', () => {
+    const resolved = resolveLogStatsWindow(
+      { minTime: '2026-01-14T06:00:00.000Z', maxTime: '2026-01-14T09:00:00.000Z' },
+      12,
+      {
+        requestedStart: new Date('2026-01-14T00:00:00.000Z'),
+        requestedEnd: new Date('2026-01-14T12:00:00.000Z'),
+        now,
+      }
+    )
+
+    expect(resolved.startTime).toEqual(new Date('2026-01-14T00:00:00.000Z'))
+    expect(resolved.segmentMs).toBe(60 * 60 * 1000)
+  })
+
+  it('reports the requested window when nothing ran inside it', () => {
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 6, {
+      requestedStart: new Date('2026-01-01T00:00:00.000Z'),
+      requestedEnd: new Date('2026-01-07T00:00:00.000Z'),
+      now,
+    })
+
+    expect(resolved.startTime).toEqual(new Date('2026-01-01T00:00:00.000Z'))
+    expect(resolved.endTime).toEqual(new Date('2026-01-07T00:00:00.000Z'))
+    expect(resolved.segmentMs).toBe(24 * 60 * 60 * 1000)
+  })
+
+  /**
+   * The case neither fallback sentence covers on its own: with no rows and only
+   * a right edge, the 24-hour window is measured back from the requested end,
+   * not from the wall clock.
+   */
+  it('measures the empty-result fallback back from a requested end', () => {
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 24, {
+      requestedEnd: new Date('2026-01-10T00:00:00.000Z'),
+      now,
+    })
+
+    expect(resolved.endTime).toEqual(new Date('2026-01-10T00:00:00.000Z'))
+    expect(resolved.startTime).toEqual(new Date('2026-01-09T00:00:00.000Z'))
+  })
+
+  /** The dashboard schema has no `startDate <= endDate` refinement, so a crossed pair reaches here. */
+  it('keeps a crossed requested pair from producing a non-positive bucket width', () => {
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 6, {
+      requestedStart: new Date('2026-01-07T00:00:00.000Z'),
+      requestedEnd: new Date('2026-01-01T00:00:00.000Z'),
+      now,
+    })
+
+    expect(resolved.segmentMs).toBe(60_000)
+  })
+
+  it('ignores an unparseable requested bound instead of stamping Invalid Date', () => {
+    const resolved = resolveLogStatsWindow({ minTime: null, maxTime: null }, 24, {
+      requestedEnd: new Date('not-a-date'),
+      now,
+    })
+
+    expect(resolved.endTime).toEqual(now)
+    expect(resolved.startTime).toEqual(new Date('2026-01-14T12:00:00.000Z'))
+  })
 })
 
 describe('buildDashboardStats', () => {

apps/sim/lib/logs/stats.ts

@@ -11,32 +11,64 @@ export interface LogStatsWindow {
   segmentMs: number
 }
 
+/** Requested window and clock overrides for {@link resolveLogStatsWindow}. */
+export interface ResolveLogStatsWindowOptions {
+  /** Caller-supplied left edge, when the request named one. */
+  requestedStart?: Date
+  /** Caller-supplied right edge, when the request named one. */
+  requestedEnd?: Date
+  /** Wall clock, injectable so the fallbacks are deterministic under test. */
+  now?: Date
+}
+
+/** An unparseable bound is treated as absent rather than as `Invalid Date`. */
+function usableBound(bound: Date | undefined): Date | undefined {
+  return bound && Number.isFinite(bound.getTime()) ? bound : undefined
+}
+
 /**
- * The window the segments span, derived from the rows that exist rather than
- * from a caller-supplied range.
+ * The window the segments span.
+ *
+ * An edge the caller named wins outright: no run outside it can be counted, so
+ * deriving the span from anything else stamps trailing buckets the query has
+ * already excluded and computes `segmentMs` over a width nobody asked for.
  *
- * A workspace with no runs still has to answer with a window, because
+ * An omitted edge still falls back to the rows that exist — the oldest matching
+ * run on the left, and on the right the later of the newest matching run and
+ * `now`, so a live dashboard's right edge is the present rather than the last
+ * thing that happened.
+ *
+ * A workspace with no matching runs still has to answer with a window, because
  * `segmentMs` and every segment timestamp are computed from one — hence the
- * trailing-24-hour fallback. The end is pushed to `now` whenever the newest run
- * is older than that, so a live dashboard's right edge is the present rather
- * than the last thing that happened.
+ * 24-hour fallback. It is measured back from the right edge, so an empty result
+ * with no bounds reports the trailing 24 hours, and an empty result with only
+ * an `endDate` reports the 24 hours preceding that date.
  */
 export function resolveLogStatsWindow(
   bounds: LogStatsBounds,
   segmentCount: number,
-  now: Date = new Date()
+  options: ResolveLogStatsWindowOptions = {}
 ): LogStatsWindow {
+  const requestedStart = usableBound(options.requestedStart)
+  const requestedEnd = usableBound(options.requestedEnd)
+  const now = options.now ?? new Date()
+
   let startTime: Date
   let endTime: Date
 
   if (!bounds.minTime || !bounds.maxTime) {
-    endTime = now
-    startTime = new Date(now.getTime() - 24 * 60 * 60 * 1000)
+    endTime = requestedEnd ?? now
+    startTime = requestedStart ?? new Date(endTime.getTime() - 24 * 60 * 60 * 1000)
   } else {
-    startTime = new Date(bounds.minTime)
-    endTime = new Date(Math.max(new Date(bounds.maxTime).getTime(), now.getTime()))
+    startTime = requestedStart ?? new Date(bounds.minTime)
+    endTime = requestedEnd ?? new Date(Math.max(new Date(bounds.maxTime).getTime(), now.getTime()))
   }
 
+  /**
+   * A crossed pair reaches here from the first-party dashboard, whose query
+   * schema carries no `startDate <= endDate` refinement, so the floor is what
+   * keeps `segmentMs` positive instead of zero or negative.
+   */
   const totalMs = Math.max(1, endTime.getTime() - startTime.getTime())
   return {
     startTime,

apps/sim/lib/workflows/application/replace-workflow-state.test.ts

@@ -11,6 +11,9 @@ const mocks = vi.hoisted(() => ({
   resolvePermission: vi.fn(),
   notify: vi.fn(),
   replace: vi.fn(),
+  prepare: vi.fn(),
+  collectGraphIds: vi.fn(),
+  assertIdsUnclaimed: vi.fn(),
   validate: vi.fn(),
   needsRedeployment: vi.fn(),
 }))
@@ -35,8 +38,13 @@ vi.mock('@/lib/workflows/application/context', () => ({
   resolveActiveWorkflowApplicationContext: mocks.resolveContext,
 }))
 vi.mock('@/lib/realtime/notify', () => ({ notifyWorkflowUpdated: mocks.notify }))
+vi.mock('@/lib/workflows/persistence/prepare-state', () => ({
+  prepareWorkflowStateForPersistence: mocks.prepare,
+}))
 vi.mock('@/lib/workflows/persistence/replace-normalized-state', () => ({
   replaceWorkflowNormalizedState: mocks.replace,
+  collectWorkflowGraphIds: mocks.collectGraphIds,
+  assertWorkflowGraphIdsUnclaimed: mocks.assertIdsUnclaimed,
 }))
 vi.mock('@/lib/workflows/sanitization/validation', () => ({
   validateWorkflowState: mocks.validate,
@@ -88,6 +96,12 @@ describe('replaceWorkflowState', () => {
       state: { blocks: { 'block-1': BLOCK }, edges: [], loops: {}, parallels: {} },
     })
     mocks.needsRedeployment.mockResolvedValue(true)
+    mocks.prepare.mockReturnValue({
+      state: { blocks: { 'block-1': BLOCK }, edges: [], loops: {}, parallels: {} },
+      warnings: [],
+    })
+    mocks.collectGraphIds.mockReturnValue({ blockIds: ['block-1'], edgeIds: [], subflowIds: [] })
+    mocks.assertIdsUnclaimed.mockResolvedValue(undefined)
   })
 
   /**
@@ -333,6 +347,77 @@ describe('replaceWorkflowState', () => {
       expect(dry.edgesCount).toBe(committed.edgesCount)
     })
 
+    /**
+     * The preview promised in {@link ReplaceWorkflowStateInput.dryRun} is
+     * byte-identical to the committed write of the same body, and preparation
+     * is where a dropped edge or a stripped inline secret is noted. Reporting
+     * only the validation half made the dry run quietly less informative than
+     * the write it previews.
+     */
+    it('merges the preparation warnings a committed write would report', async () => {
+      mocks.validate.mockReturnValue({
+        valid: true,
+        errors: [],
+        warnings: ['Dropped block "block-2"'],
+      })
+      mocks.prepare.mockReturnValue({
+        state: { blocks: { 'block-1': BLOCK }, edges: [], loops: {}, parallels: {} },
+        warnings: ['Dropped edge "edge-9": target block does not exist'],
+      })
+      mocks.replace.mockResolvedValue({
+        warnings: ['Dropped edge "edge-9": target block does not exist'],
+        state: { blocks: { 'block-1': BLOCK }, edges: [], loops: {}, parallels: {} },
+      })
+
+      const dry = await replaceWorkflowState.execute({
+        principal: sessionPrincipal,
+        input: { ...input, dryRun: true },
+      })
+      const committed = await replaceWorkflowState.execute({ principal: sessionPrincipal, input })
+
+      expect(dry.warnings).toEqual([
+        'Dropped block "block-2"',
+        'Dropped edge "edge-9": target block does not exist',
+      ])
+      expect(dry.warnings).toEqual(committed.warnings)
+    })
+
+    /**
+     * A dry run that reports clean for a body that cannot commit is worse than
+     * the fault it hides. It checks the ids the write would actually insert —
+     * the prepared graph's, not the caller's body's.
+     */
+    it('refuses a graph whose ids another workflow already owns', async () => {
+      mocks.assertIdsUnclaimed.mockRejectedValueOnce(
+        new OrchestrationError('conflict', 'Block ids already used by another workflow: block-1')
+      )
+
+      await expect(
+        replaceWorkflowState.execute({
+          principal: sessionPrincipal,
+          input: { ...input, dryRun: true },
+        })
+      ).rejects.toMatchObject({ code: 'conflict' })
+      expect(mocks.replace).not.toHaveBeenCalled()
+    })
+
+    it('checks the ids the prepared graph would insert, not the ids sent', async () => {
+      const prepared = { blocks: { 'block-1': BLOCK }, edges: [], loops: {}, parallels: {} }
+      mocks.prepare.mockReturnValue({ state: prepared, warnings: [] })
+
+      await replaceWorkflowState.execute({
+        principal: sessionPrincipal,
+        input: { ...input, dryRun: true },
+      })
+
+      expect(mocks.collectGraphIds).toHaveBeenCalledWith(prepared)
+      expect(mocks.assertIdsUnclaimed).toHaveBeenCalledWith(expect.anything(), 'workflow-1', {
+        blockIds: ['block-1'],
+        edgeIds: [],
+        subflowIds: [],
+      })
+    })
+
     /** A locked workflow refuses the preview too, or the preview would lie. */
     it('refuses when the workflow cannot be mutated', async () => {
       workflowAuthzMockFns.mockAssertWorkflowMutable.mockRejectedValueOnce(