From 4a198653f11487fba3896eedc4f1202f63021f5e Mon Sep 17 00:00:00 2001 From: Justin Blumencranz <96924014+j15z@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:27:30 -0700 Subject: [PATCH 1/6] fix(docs): hide unreleased table expiration (#7401) --- apps/docs/content/docs/tables/index.mdx | 3 - apps/docs/openapi-v2-tables.json | 90 +++---------------------- apps/sim/lib/api/contracts/tables.ts | 4 +- scripts/openapi/documents.test.ts | 4 ++ scripts/openapi/generator.test.ts | 29 ++++++++ scripts/openapi/generator.ts | 37 ++++++++-- 6 files changed, 75 insertions(+), 92 deletions(-) diff --git a/apps/docs/content/docs/tables/index.mdx b/apps/docs/content/docs/tables/index.mdx index 41b821d995e..e44ca483464 100644 --- a/apps/docs/content/docs/tables/index.mdx +++ b/apps/docs/content/docs/tables/index.mdx @@ -24,7 +24,6 @@ Every column has a type, which decides how its values are stored and validated. | **Currency** | An amount in a currency you pick per column | `$1,234.56` | | **Boolean** | `true` or `false` | `true` | | **Date** | A date | `2026-03-16` | -| **Expiration** | An absolute row expiration time, stored as Unix epoch seconds (seconds since January 1, 1970 UTC) | `1773671400` | | **JSON** | An object or array | `{ "tier": "pro" }` | | **Select** | One of a fixed set of options, or several | `Pro` | @@ -32,8 +31,6 @@ Types are enforced as you enter values, so a Number column only takes numbers. A Currency column stores a plain number and renders it in the currency you choose for that column, so filters, sorts, and exports all see the amount itself. Changing a column's currency relabels it — it does not convert the amounts. -A table can have one Expiration column. Adding it enables row expiration; rows with a non-empty expiration value become eligible for deletion after that time passes. Cleanup runs periodically, so actual row removal may happen after the expiration timestamp rather than exactly at it. Deleting the Expiration column disables expiration for the table. Expiration cells use the date editor, while APIs and workflows read and write integer Unix epoch seconds. - ## Editing a table Open the **Tables** section in the sidebar and click **New table** to create one. Add columns from the column header, type into a cell to edit it, and paste rows from a spreadsheet to bulk-load. Filter and sort from the toolbar without changing the underlying data. The editor has full keyboard support; see [keyboard shortcuts](/keyboard-shortcuts). diff --git a/apps/docs/openapi-v2-tables.json b/apps/docs/openapi-v2-tables.json index d91748c097d..fb9688e5f6c 100644 --- a/apps/docs/openapi-v2-tables.json +++ b/apps/docs/openapi-v2-tables.json @@ -4979,16 +4979,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -5266,16 +5257,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Column data type." }, "required": { @@ -5454,16 +5436,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -5563,16 +5536,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Column data type." }, "required": { @@ -5669,7 +5633,7 @@ "type": { "description": "Replacement column data type.", "type": "string", - "enum": ["string", "number", "currency", "boolean", "date", "ttl", "json", "select"] + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"] }, "required": { "description": "Whether inserts must supply a value for this column.", @@ -7433,16 +7397,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -7642,16 +7597,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Output column data type." }, "required": { @@ -7792,16 +7738,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Output column data type." }, "required": { @@ -7919,16 +7856,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { diff --git a/apps/sim/lib/api/contracts/tables.ts b/apps/sim/lib/api/contracts/tables.ts index 9d3a95e206e..30dd03797e3 100644 --- a/apps/sim/lib/api/contracts/tables.ts +++ b/apps/sim/lib/api/contracts/tables.ts @@ -53,7 +53,9 @@ export const domainObjectSchema = () => z.custom(isRecordLike) * Column types are a fixed enum derived from `COLUMN_TYPES` so callers cannot * send arbitrary strings the server would reject downstream. */ -export const columnTypeSchema = z.enum(COLUMN_TYPES) +export const columnTypeSchema = z + .enum(COLUMN_TYPES) + .meta({ omitEnumValuesFromOpenApi: ['ttl'] as const }) /** One choice in a `select` column. `id` is the stable cell key. */ export const selectOptionSchema = z.object({ diff --git a/scripts/openapi/documents.test.ts b/scripts/openapi/documents.test.ts index d2c31bd694b..c7c3b510584 100644 --- a/scripts/openapi/documents.test.ts +++ b/scripts/openapi/documents.test.ts @@ -279,6 +279,10 @@ describe('generated OpenAPI documents', () => { expect(tableProperties.ownerEmail).toMatchObject({ type: 'string', format: 'email' }) }) + it('omits feature-flagged table column types', () => { + expect(JSON.stringify(generatedDocument(tablesOpenApiDocument))).not.toContain('"ttl"') + }) + it('keeps billing as its own API reference group', () => { const spec = generatedDocument(billingOpenApiDocument) expect((spec.tags as JsonObject[]).map((tag) => tag.name)).toEqual(['Billing']) diff --git a/scripts/openapi/generator.test.ts b/scripts/openapi/generator.test.ts index b6b9b453fb9..14fbd811396 100644 --- a/scripts/openapi/generator.test.ts +++ b/scripts/openapi/generator.test.ts @@ -194,6 +194,35 @@ describe('OpenAPI generator', () => { }) }) + it('omits feature-flagged enum values from generated schemas', () => { + const columnType = z.enum(['string', 'ttl']).meta({ omitEnumValuesFromOpenApi: ['ttl'] }) + const body = z + .object({ type: columnType.describe('Column data type.') }) + .meta({ id: 'HiddenEnumRequest', title: 'Hidden enum request', description: 'Request body.' }) + const response = z + .object({ ok: z.boolean().describe('Whether the request succeeded.') }) + .meta({ id: 'HiddenEnumResponse', title: 'Hidden enum response', description: 'Response.' }) + const contract = defineRouteContract({ + method: 'POST', + path: '/hidden-enum', + body, + response: { mode: 'json', schema: response }, + }) + const route = defineOpenApiRoute( + contract, + operation('hiddenEnum', { description: 'Response.' }), + { body, response } + ) + const spec = generateOpenApiDocument(document([route])) + const schemas = (spec.components as JsonObject).schemas as JsonObject + const requestProperties = (schemas.HiddenEnumRequest as JsonObject).properties as JsonObject + const documentedColumnType = requestProperties.type as JsonObject + + expect(columnType.safeParse('ttl').success).toBe(true) + expect(documentedColumnType.enum).toEqual(['string']) + expect(documentedColumnType).not.toHaveProperty('omitEnumValuesFromOpenApi') + }) + it('handles every route response mode and media type', () => { const emptyContract = defineRouteContract({ method: 'DELETE', diff --git a/scripts/openapi/generator.ts b/scripts/openapi/generator.ts index 550113a1ce8..93fb3bb762b 100644 --- a/scripts/openapi/generator.ts +++ b/scripts/openapi/generator.ts @@ -105,6 +105,31 @@ function stripLegacySchemaIds(value: unknown): unknown { ) } +function omitEnumValuesFromOpenApi( + metadata: z.core.GlobalMeta | undefined, + schema: JsonObject, + label: string +): void { + const omittedValues = metadata?.omitEnumValuesFromOpenApi + if (omittedValues === undefined) return + + invariant( + Array.isArray(omittedValues) && omittedValues.length > 0, + `${label} omitEnumValuesFromOpenApi must be a non-empty array` + ) + invariant( + Array.isArray(schema.enum), + `${label} omitEnumValuesFromOpenApi requires an enum schema` + ) + for (const value of omittedValues) { + invariant(schema.enum.includes(value), `${label} omits an enum value that does not exist`) + } + + schema.enum = schema.enum.filter((value) => !omittedValues.includes(value)) + invariant(schema.enum.length > 0, `${label} cannot omit every enum value`) + Reflect.deleteProperty(schema, 'omitEnumValuesFromOpenApi') +} + function comparableSchema(schema: ApiSchema, io: SchemaIo): unknown { const cached = comparableSchemaCache.get(schema)?.get(io) if (cached) return cached @@ -206,14 +231,12 @@ function generateSchema( unrepresentable: 'any', cycles: 'ref', reused: 'inline', - override: ({ zodSchema, path }) => { + override: ({ zodSchema, jsonSchema, path }) => { const current = zodSchema as ApiSchema - validateExamples( - current, - z.globalRegistry.get(current)?.examples, - io, - `${label} at ${path.join('.') || ''}` - ) + const metadata = z.globalRegistry.get(current) + const schemaLabel = `${label} at ${path.join('.') || ''}` + validateExamples(current, metadata?.examples, io, schemaLabel) + omitEnumValuesFromOpenApi(metadata, jsonSchema as JsonObject, schemaLabel) }, }) as JsonObject const byIo = generatedSchemaCache.get(schema) ?? new Map() From d6e083ecbd685b5a4bb76ea908b74755253ae92a Mon Sep 17 00:00:00 2001 From: Waleed Date: Wed, 2 Sep 2026 11:33:24 -0700 Subject: [PATCH 2/6] feat(providers): add Gemini 3.8 Flash (#7402) Claude-Session: https://claude.ai/code/session_01LKqTj3FNr4iS5jMy1gZe2D Co-authored-by: Claude Opus 5 (1M context) --- .../content/docs/workflows/blocks/agent.mdx | 2 +- apps/sim/providers/models.ts | 21 ++++++++++++++++++- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/apps/docs/content/docs/workflows/blocks/agent.mdx b/apps/docs/content/docs/workflows/blocks/agent.mdx index 960bd7bab02..6bb78780dd7 100644 --- a/apps/docs/content/docs/workflows/blocks/agent.mdx +++ b/apps/docs/content/docs/workflows/blocks/agent.mdx @@ -112,7 +112,7 @@ Live tool-call chips stream for **OpenAI, Anthropic, Azure Anthropic, Google, Ve | Anthropic | Summaries only — These generations omit full thinking; Sim requests summarized thinking on streaming runs. | `claude-fable-5-1`, `claude-fable-5`, `claude-sonnet-5`, `claude-opus-5`, `claude-opus-4-8`, `claude-opus-4-7`, `claude-opus-4-6`, `claude-sonnet-4-6`, `claude-opus-4-5`, `claude-opus-4-1`, `claude-sonnet-4-5`, `claude-haiku-4-5` | | Azure OpenAI | Summaries only — Requires OpenAI organization verification; falls back to no summaries. | `azure/gpt-5.4`, `azure/gpt-5.4-mini`, `azure/gpt-5.4-nano`, `azure/gpt-5.2`, `azure/gpt-5.1`, `azure/gpt-5.1-codex`, `azure/gpt-5`, `azure/gpt-5-mini`, `azure/gpt-5-nano`, `azure/o3`, `azure/o4-mini` | | Azure Anthropic | Summaries only — These generations omit full thinking; Sim requests summarized thinking on streaming runs. | `azure-anthropic/claude-opus-4-6`, `azure-anthropic/claude-opus-4-5`, `azure-anthropic/claude-sonnet-4-5`, `azure-anthropic/claude-opus-4-1`, `azure-anthropic/claude-haiku-4-5` | -| Google | Summaries only | `gemini-3.6-flash`, `gemini-3.5-flash-lite`, `gemini-3.5-flash`, `gemini-3.1-pro-preview`, `gemini-3.1-flash-lite`, `gemini-3-flash-preview`, `gemini-2.5-pro`, `gemini-2.5-flash`, `gemini-2.5-flash-lite` | +| Google | Summaries only | `gemini-3.8-flash`, `gemini-3.6-flash`, `gemini-3.5-flash-lite`, `gemini-3.5-flash`, `gemini-3.1-pro-preview`, `gemini-3.1-flash-lite`, `gemini-3-flash-preview`, `gemini-2.5-pro`, `gemini-2.5-flash`, `gemini-2.5-flash-lite` | | Vertex AI | Summaries only | `vertex/gemini-3.5-flash`, `vertex/gemini-3.1-pro-preview`, `vertex/gemini-3.1-flash-lite`, `vertex/gemini-3-flash-preview`, `vertex/gemini-2.5-pro`, `vertex/gemini-2.5-flash`, `vertex/gemini-2.5-flash-lite` | | DeepSeek | Full thinking deltas | `deepseek-v4-pro`, `deepseek-v4-flash`, `deepseek-reasoner` | | xAI | Full thinking deltas | `grok-4.6`, `grok-4.5`, `grok-4.3`, `grok-4.20-multi-agent-0309` | diff --git a/apps/sim/providers/models.ts b/apps/sim/providers/models.ts index 83767ac68ef..1b09f47db67 100644 --- a/apps/sim/providers/models.ts +++ b/apps/sim/providers/models.ts @@ -1594,6 +1594,26 @@ export const PROVIDER_DEFINITIONS: Record = { icon: GeminiIcon, color: '#4285F4', models: [ + { + id: 'gemini-3.8-flash', + pricing: { + input: 0.75, + cachedInput: 0.075, + output: 3.75, + updatedAt: '2026-09-02', + }, + capabilities: { + temperature: { min: 0, max: 2 }, + thinking: { + levels: ['low', 'medium', 'high'], + default: 'medium', + }, + maxOutputTokens: 65536, + }, + contextWindow: 1048576, + releaseDate: '2026-09-02', + recommended: true, + }, { id: 'gemini-3.6-flash', pricing: { @@ -1612,7 +1632,6 @@ export const PROVIDER_DEFINITIONS: Record = { }, contextWindow: 1048576, releaseDate: '2026-07-21', - recommended: true, }, { id: 'gemini-3.5-flash-lite', From 27ca32b9cb46c0ff6235e3b45608c4b284bb6776 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 2 Sep 2026 11:48:33 -0700 Subject: [PATCH 3/6] feat(library): What Is Human-in-the-Loop in AI Agents? (#7403) Co-authored-by: Sim Pi Agent --- .../index.mdx | 82 ++++++++++++++++++ .../cover.jpg | Bin 0 -> 23295 bytes 2 files changed, 82 insertions(+) create mode 100644 apps/sim/content/library/what-is-human-in-the-loop-in-ai-agents/index.mdx create mode 100644 apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg diff --git a/apps/sim/content/library/what-is-human-in-the-loop-in-ai-agents/index.mdx b/apps/sim/content/library/what-is-human-in-the-loop-in-ai-agents/index.mdx new file mode 100644 index 00000000000..096e9756de5 --- /dev/null +++ b/apps/sim/content/library/what-is-human-in-the-loop-in-ai-agents/index.mdx @@ -0,0 +1,82 @@ +--- +slug: what-is-human-in-the-loop-in-ai-agents +title: 'What Is Human-in-the-Loop in AI Agents?' +description: 'Learn how human-in-the-loop AI agents use approval gates, in-line review, exception escalation, and post-hoc audits to control consequential actions.' +date: 2026-09-02 +updated: 2026-09-02 +authors: + - andrew +readingTime: 7 +tags: [AI Agents, Human in the Loop, Workflow Automation, Sim] +ogImage: /library/what-is-human-in-the-loop-in-ai-agents/cover.jpg +canonical: https://www.sim.ai/library/what-is-human-in-the-loop-in-ai-agents +draft: false +faq: + - q: "How does agentic HITL differ from HITL in model training?" + a: "Agentic HITL governs runtime decisions, while training HITL uses human feedback during model development or evaluation. Sim places human review within workflow execution rather than model training. Runtime review can stop a questionable action before it affects another system." + - q: "Does HITL slow down AI agents?" + a: "Human review adds waiting time wherever a workflow pauses for a decision. In Sim, you can reserve the Human in the Loop block for actions that require judgment. Selective checkpoints preserve speed for low-risk work while protecting sensitive actions." + - q: "How should approval timeouts and escalations work?" + a: "In Sim, a Human in the Loop block has no default timeout. It pauses indefinitely until someone responds through the approval portal, the API, or a webhook. If a workflow needs a hard deadline, pair the block with a separate timeout mechanism and route expired requests to the safer outcome, such as rejecting the action or escalating to another reviewer." + - q: "Does every agent action need a human checkpoint?" + a: "A checkpoint controls a specific action, so an agent does not need one for every step. In Sim, you can place human review at selected points in a workflow rather than applying it to every action. Read-only retrieval and low-impact drafting may run autonomously, while irreversible or regulated actions should receive review." +--- + +## TL;DR + +- Human-in-the-loop AI places human approval, review, or intervention at selected points while an AI agent runs. +- AI agents need more oversight than deterministic automation because probabilistic decisions can produce incorrect outputs or tool calls, even when the workflow runs as designed. +- Four common HITL patterns are approval before execution, in-line review, exception-based escalation, and post-hoc audit. +- Full autonomy generally suits low-risk, reversible work. Human checkpoints are better suited to consequential actions that are difficult to reverse or require regulatory review. + +## Why agentic workflows need human checkpoints + +AI agents need checkpoints during execution because language models can choose actions from uncertain or incorrect premises. An agent may misread a request or select the wrong tool based on an invented fact. Rule-based automation can also contain bugs, but the same input generally follows the same programmed path. Agent behavior can vary even when inputs look similar. This distinction is central to understanding [how agentic workflows differ from traditional automation](https://www.sim.ai/library/what-is-an-agentic-workflow). + +Small error rates become significant when an agent performs many actions. If an agent completes each action correctly 99 percent of the time, it has about a 63 percent chance of making at least one error across 100 independent actions. This calculation assumes that each action has the same error rate and that the outcomes are independent. Repeated execution gives individual errors more opportunities to affect a customer message or account record. Earlier mistakes may also affect later decisions when the agent treats a false output as reliable context. + +Human checkpoints let a reviewer inspect an agent's proposed decision before it affects an external system. An approval gate can stop a mistaken tool call before the agent sends an email, issues a refund, or changes a database record. Separating the proposal from execution gives the reviewer a chance to catch an invented claim or incorrect tool call before the agent changes the external system. + +You can place checkpoints around every sensitive action, trigger them only for exceptions, or audit low-risk actions afterward. Choose among these patterns based on reviewer capacity, the potential cost of an error, and whether the action can be reversed. + +## Four human-review patterns for agent workflows + +Four common HITL patterns place human judgment at different points in an agent's run. Earlier checkpoints can prevent an agent from completing a harmful action, but they add delay and reviewer workload. Selective or post-hoc review requires less immediate attention, although reviewers may not see an error until after it affects another system. + +**Approval-before-execution requires a decision before the agent acts.** The agent prepares a structured action, such as a refund or database update, and pauses while a person approves or rejects it. A propose-versus-commit design keeps preparation separate from execution. If the reviewer rejects the proposal, the agent does not create the side effect. This pattern fits infrequent actions with high cost or limited reversibility. + +**In-line review lets a person inspect and edit an output before release.** An agent might draft an email or report, then send the draft to a reviewer who can revise, approve, or reject it. The reviewer controls the final content while the agent produces the first draft. In-line review works well when judgment affects wording, factual accuracy, or policy compliance. + +**Exception-based escalation allows routine actions to proceed automatically.** The workflow pauses only when a defined trigger fires, such as low model confidence or a transaction above a set value. Rules can also require review whenever an action handles sensitive data. You must set escalation thresholds and assign a reviewer. You should also define what the workflow does when nobody responds. This pattern reduces approval volume, but poor trigger rules can let risky cases bypass review or send too many safe cases to humans. + +**Post-hoc audit reviews actions after the agent completes them.** A person may inspect sampled outputs or investigate flagged runs, then correct mistakes that remain reversible. Since the agent does not wait for approval, post-hoc audit suits high-volume actions whose failures have limited impact. The pattern provides less protection because reviewers encounter bad actions only after they occur. + +You can combine these patterns within one workflow. A payment agent might require approval above a value threshold and audit a sample of lower-value transactions. Separate exception rules can escalate unusual payments at any value. Assign each action a checkpoint based on the likely harm and whether that harm can be reversed. Reviewer workload should then determine whether eligible actions receive universal or sampled review. + +## When to let the agent run and when to stop it + +Choose the level of autonomy by comparing the evidence supporting an agent's proposed action with the consequences of an incorrect decision. Full autonomy generally fits routine actions with limited consequences and performance that you have validated through testing. Examples include read-only retrieval or internal logging. Record classification may also qualify when errors are easy to detect and correct. + +Irreversible actions need approval before execution because correction may be impossible or expensive. Examples include deleting records, changing production data, or sending payments. Refunds may also require approval when they are costly or difficult to reverse. Actions with serious or widespread consequences should require approval before execution even when the agent reports high confidence. + +High-stakes outputs that remain editable fit an in-line review checkpoint. A person can revise a customer email or public post before publication. The same review can apply to a report before distribution. Compliance-sensitive steps often need approval before execution and a post-hoc audit record that identifies the proposal, reviewer, decision, and final action. + +Low-confidence outputs fit exception-based escalation. You can set a confidence threshold and route uncertain cases to a reviewer while allowing routine cases to continue. Confidence scores may not correspond reliably to actual error rates, so do not use them as the sole escalation trigger. Add rules for sensitive data, unusually valuable transactions, or evidence that conflicts with the proposed action. + +A supervised rollout provides evidence for setting those thresholds. Start by reviewing every consequential action. Measure how often reviewers reject proposals and record the error types behind those rejections. Once observed performance supports narrower review, move reliable actions to exception-only escalation or sampled post-hoc audits. Keep approval gates when the consequences of a single incorrect action remain unacceptable regardless of past accuracy. + +## How Sim's workflow builder handles human review steps + +Sim exposes a dedicated [Human in the Loop block](https://docs.sim.ai/workflows/blocks/human-in-the-loop) within its workflow builder. Sim lists the Human in the Loop block alongside Guardrails and Evaluator blocks as separate core blocks, and a separate Wait block exists too, so a human review checkpoint is a distinct primitive from a generic delay. + +The block pauses a run and waits for a person before continuing. Configuration covers what the approver sees, drawn from earlier block outputs, how they get notified through Slack, Gmail, Microsoft Teams, SMS, or a custom webhook, and a resume form that captures their decision. By default the run stays paused with no timeout until someone responds through the approval portal, the API, or a webhook, and downstream blocks can read the approver's inputs directly. Workflows can chain multiple Human in the Loop blocks for staged approvals, such as a manager sign-off followed by a director sign-off. + +Sim's logs record every run block by block, so a reviewer can inspect the path a workflow took, see what each block produced, and confirm whether a review step ran. That record supports the post-hoc audit pattern even on workflows that never pause for approval, and it helps identify where future runs should add a checkpoint. For a broader treatment of logs and traces, see this guide to [AI agent observability](https://www.sim.ai/library/ai-agent-observability). + +## HITL as one piece of a larger agentic workflow + +Human review complements the other controls in an agent workflow, including retrieval and restricted tool access. An agent may use retrieved information to prepare a response or proposed tool call. The workflow can then pause for approval before the external tool changes another system. Each workflow step can use a different level of oversight. + +Retrieval-augmented generation, commonly called RAG, grounds an agent's response in fetched documents or data. Grounding a response in relevant sources can reduce unsupported claims when retrieval returns accurate material and the model uses it correctly. Retrieval does not prevent every mistaken interpretation or unsafe tool call. Evaluators and guardrails can flag conditions they are configured to detect. A person can review cases that require contextual or policy judgment. Read more about [how retrieval-augmented generation works](https://www.sim.ai/library/what-is-retrieval-augmented-generation). + +Match each action to the least restrictive control that keeps its consequences acceptable. Read-only retrieval may run autonomously, while a costly refund may require approval and reversible actions may receive sampled review after execution. Sim represents these checkpoints through its [Human in the Loop block](https://docs.sim.ai/workflows/blocks/human-in-the-loop). Review the block in Sim's workflow builder to determine where a human decision adds useful control without delaying routine work, then follow the practical guide to [creating an AI agent](https://www.sim.ai/library/how-to-create-an-ai-agent) when you are ready to build the workflow. diff --git a/apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg b/apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg new file mode 100644 index 0000000000000000000000000000000000000000..44fe6f1148a76d5c8f65d89d5fb51bdf38e7cfda GIT binary patch literal 23295 zcmeFYbwFH8)-T!wcL@^Q-3jgxG4Ej%p)-T)B(JP;8PkWru?bTm{{G;~}HOy~!f z{3ZTN=$D+Bl9&*Ba?(*z(Q)##v$FF`iHJyPsA)JjBtJI-^MBk(4*(nKMIKxgEDROk z1vU&UHq28WfCK;ocmWFo0|WT$KtP0pM}mFv{1o>e^ZzXHv;;tdh1S7>#e!zkcl}lE zKTg1lnGk>Zk84RAN>3ERPe34(0630St1?D4BlC;z8xj^g=(+ZyH_Pri#6;|9dp0PK zbb2~uu37m0|5+E)GP~7WsF=eEsM)E_T>zj)1>}=fV#k+;E1vOOwYhO6n?&Hn8<+4y zl@+|4{N^$RmYKUmS%H-qvb5s<07p8X(!m9MzVFOK+lB9TwX{(7K3Eoq*GkjfBANT5bKU4 z{Nq0&nMcUzRf7M2ROSB`{=Xdf{|5(x$El>426Aj_MiCk+6J~;YdGbyGRuQ{&ty#&g zK#PlYr!i{4?RB`>q9`J3EWRZ&vUsU|xBM3%EL$e-7FN2?VvcwK0 zG*{W9ntD}w`zwJDoPA|P$0SeY&7}8lPf(M_H!Y0^W@+-qBEes3y0Z_Ic~q^7bpEB6 zQw-Hps{5jj<9E@zi7BH_T__bZr6X6Dxs~tP{M$c(?M2INUdrpQt(h0O|9dn6$c2x; zxhz7KdYzQYk3m}7&RY6)0v~`k9q`li=a+wyq~)e(X%?Y;lt1(Yka7MjL*$t;s%v-Rd-`&Ll&9e6wlus2q^Gw^TH zge9<3wcESUlE3O3x+~-Y@Gu5=We<+)o2hOwd>)`lW6PVG(y%fSE zrujF?ft9~*0&qBo#9Dv>7t<#Q7!EvM68Yah0qe0zvyY%=zHlXUB2bcQ`Th&6K^2im zW0m&<(p7^wHJJR%zuCqA=VF=*1<3$(Zi+hW;aC6M2pTny=FsqmKQobjiPsUefBEx< z>3mXr*ZZ>M**-=jCTuB-}O}5P(Ta7ocjR0KZ^rkclQ7Q%+Fr{0LyDK z3*LqIW3}7>axrlL08!|K+pFgY4yCZc%J?)v#JI=26cse*C+Oq)`~U!8 zw{D>gBtTmcRe`qh*MZB{XjUvJtr;ny8II1$5t>{OtiK@5G#|N#RXQ+G5;^Zh8kdOi zNJQhf=-2$Eu0)M7$}WG4X}sXE`OpH$+Ra9lg2@gifulf_=g&L zEaU0S3;Q>O{xB5_hbW?6nhX6o$}8#~(VYQ5$7LZuaX@nfJT5Tx02He0ijmIgYyO+2 z+}c3q_c}bBB4HlNNN|lfye@w2&L-qZBh{}tc9(l#tY@|J@<1N`|2Qru|B$0ATL)tjyT;+# zYQI|YIeWR*{iR?Zxvb-Bq-RcJct5I-YGG?mvGs3_IYp$BN+#aZ*PEjC$Tp=$-rH~J zEJQ)G<5AkUKI^jat6^U}Da0+ly-00g^vYb7b}k8ARPVWro4pE{xEm)sZH0MJQl6gL zXu+Ar89tH4I!-gUR&*@+X=Nbz{)=?Ka&SHUC%Y&!Mb@lKY_nto?q7}EVC$aPHmFo+ zZ+u!AwBedYAP{j$0@$;Zx{WesptLuIf1ye&lD^2XQeswM+4azUH^S0oEpS&Lu$Q3o z3wZEf9QjvSqzb+njU}b|W*(1{D?$D@o@`)3d7^gfGmyFGuUW5G0!)F}$-`45^EiJ) zz!#3FW`^>lA$%LklcKB#@wFa!kpMiNb=g(8$Uz33dcHHYct1L{6Rku7Y520*MzWF7CWmbILp(uF?i|5<@LOYEnLhzlXcXR3D(kP zycxghw@5dDu;gidy-4LWUhs@=p14ByNp*QI+e~t3G-q0PT*<=mUrb@pvmM`10KiG6 z?z!JUs_J##5i0+JxaTG(&y<`BqRN>Ld2_{BXDVFL{l@jp-X5eithDL~m`YpKX{*if zVRvuVdRW4l&L$X0Ojm83F)caSYx^Qz)YAKz9nF$36+krpIeh$zJJiPR!(9*mdp&I2 zq_Ch)z;0v~#p^BrZkudAl^LJC$B10H(xm5WI;GH5sCwtZacxZ6k4sZeb|CWK>F~qr z(N)vR4i*Do(>=8*-lJwdfAJ^WPA~EWiC^ssM;ZXi3h+ttZ+;;q2}}`fe)eHy$AMJS zd7XXA3UuzU!wb^v)C_VjYrG=qy1DwVvHu4t!(F4yLn9gUf{6d1(*IcW1@@u(F2UsU zbJ%m_`g~M#iwU=8F<5eHkhEJ})MS%C0LeQg8&yL|Txrz%uV!8puofWu_(Ks4+`BrK zN}}t%MswjsT6TiLg}=)D_XPImCIm+-EGLJ&9eDqo>WKi2WVKrW0;l>L(QPqfR&djfXonuUxf%~Igr z=FxKNM_z!KQTm$kD> z$%&@Lo|++Ks25nyiETHBYTMQn0XG{3;M)mPZb$WD#_t8^L0NV4+>4#E-^&|j^ZoXl z>yGhJLaEMg9TC})S*-WUH+^}lOK$20`6oJz5are9lC;HZ94m1;V_c*Z3+$PAzlxMi z1QjCXPf>ph()p>Rk7wFJX;1Xq_4wSmQ^L^B|J7x^@7wh!fCeSw|I32xA4W6v_{DzH z@mU<(pFL4AfcqC-q%K(gQjm?oMwrdhSXYl|JiBUlIG^o?O?ALY z(B(kao>lf7WMnPZW9gR9KXAs))>NNJWampsq)Ty1ac)ZSQ01{;Re+0yz%Qeuq7};# z1n7dbX0(va(=hru({ScERLIE1MZMTmr#Qe$;imM4&%fC4S|c&t#{Vzw|4U+bJiFFh zAFm)q8?OLr>IOCuh2VC*P}-{^RWUyor@h5)2B?0zzHoi@gcwyzs>UUS0n(h`%xdx5gwzDDjTgScH9oz1qS4*d+{==8OISQ3_t9 z(BX4wuf`;QBA>2GGiXo*Noqyh+`|zTMP7U)lEAuf{uNKQ{!HUXvCp}j^mqo_rBqH` zGBg|QpFI)iK$WIkSZ7yQtY%-Rrs86yAnWh#Sg>Q`0FqS@%HZ_B2>9P6;Qe%6_;)^c zL|ihCnh5{zO8*}{e-172VgX>_U|zr?!otEq<4Ldp!1EnE0ssq8 zKiloS3dT?BqfD|L$O5l>M1G<$wUeYrwayR zlk8NN?zTK8#r%Z4?s|hG$E{T0VD}R&(ypZ}BXM&)%R=BF4fh1FKjevj|NdP)Wh6*e zKlU?cTRH?3N|YXUxL+Qg1vqs!%m=* z4G_f>facuQIi?{wRo=Wu@W}gKB7E-y&LKi0(g&m$K03y;^R$#w>a&>2OcA0A^EQ_e z_r8rbZlz46-3#9>@Fqjekg?X296?&5T9+U7&($ON zpWS}Xc}Q*8pg+A`U%Mdxyw%L2e#V032lfygs4gzqkl&sClXsW=sp>nF4i}(p*CewM zO1!~ta}w!T8nv;0)8-_(q)_p72W0(&O)A}#ip4{`-w$?^k7GKHFMsEr$2ZnMbw6e^MQqx{w((LiodIQ{2K^nQj3&!L?kiP27_^V4&{+#4m* z5MZ|M_NYD@FsF+~)W2w2JIY!+z4;5q#i&QR6=3iFYx43W?h_zO_kE-bQ@02P2P;(v z@Coo+vZi%ZP%~uR=i^6++qbJrha|RnL{?j)f})C_ml3AYx+A8vKQ)}KrsU+9oDN%w zDf(ukPVn0{s4u^-Rtrtr-9G`+A2fBqkw^5I5*^cxHz4f>J-&aiMCg4p~P8jLA%x_aL--h&!!eo-$MC92|!WUMA25{dLhvtv4F?+inJIq_Q?d{y*p{bVa_wLjwJYjYvsaEXs)=fyIC|Q!!n`SDd zaq54qc4lYa#txFzi_HiVJ@2&9+`McZ<>uh)sl>UJq<}F7+Pu%yt!!f zmuYF)yIPjUO^_SPt?a@(tG~kM$8;hX)-NsxO7|-{Xa+D@3pIG#gw(LUGs0FBHq{yp zQj_bqX4~H>;E{qrKJ^&QQdaK3a_V9p8y|Q%WQB6IL~LJwJl|CUKKND$YG}3 zmoZ>YsAc!fV@4ijp3aN1_f+7jTkBr)l*5?^Vde30V2EW=<3Ku~D!%-+;l-{Et$kM#C9Ss4w|W({xkf5=Z3P)+E@TuDfqj9rQz zKKaq^9*I#VXzTkg(g!I-jF&JSuqGYe$}^1<)*T7Uo;ck>(*77?cq+?XTYFx_Q)!*P z8R8(5__{jqKI;dj?p7z+Hie?Dn1DMhfv&{M1}rqWEthw))UU1Sl-_F$U4|iUGXw+w zpfS{M%s$=Qza0JqZfR%lty-JqQB7+kp0i771rQiPSoG0}s%!$jn!jAbW$7XMI`|Oj zu=>IgaaHs9*=W8ED=uzq(o!9z!~;vSu)8?z$9mhz)|7G8!5ron7Vsy+ddEMRWD zlG2EFImewIEEp|ii?ek$!&&iHjB5@#An^}+0w84U0N4u z!_7c(0n)7PJt)1!^2h2;-K$vBxMWS<3g7X<%5F5g$s=_+>J(x#lDqVgIf1@4 z-C;MUxZ`(ZguwcJ{1|D=LaNPV)ma^J#iOC()fp!g!$HXe^iC|FZQq5;!9&*sGi23L zie8;;#(l$)%!Z2LXq$2`-5F;ml882}h~V_~-?4^+$tNQfQWvwb(N(9yV2WqHP({$$ zW`_UJ{k>m79{T;&CV@3ny&Ex}0I6%-H!#|$tH=!o(QlnQzTo8611lD6j2B98CaiLE zuh_`X1&?b{w0f#D2W*2}7;gCgZr92VuqzQ|IwsNw3Z;#7rc4S~MrJ<*iX%-%BW_zq ze`6|zaO3ull1%=zHN+Ps&-gWqAB<&~BZsgA%1dXYN|;vHMyW$Nc{ySOq*bdGbkd1Y zs~-1q69ktI1N%LxLM1j6zciC0MCKI>98$h8`SnA&2VpvWzJA=}RXGe(g_BX!IlEN| z))(F`r_*WV6U!> zk76*UWJ3!3^tN~gs57?K<26{y)OV3t!HB1(rivD^(L4?+=Hj&5&?wq7dIC^a>;PqQ zo9EMds!zy@%U+HF8OcaU;dO1{KQ1G{tnrVp}L1m5~*=~g7Ifehm#!b$o>M< z_c=2=jYjP9;HRGoHQ~$g81Hhm?bbfC`rJeO;)2`)ysXv!{SwM*)nglRCw|?J>@mYf zSZmic`mGklrl;l+dOkfF1`dVuP`SNEvDGu1(hXcEtswUjLVqp-v96f2@54yPU+^1b zuID=%Z#O3}_f|@`CL!wLC+E0(zd1+nwp2J%l#v=3RgDA}E<!1pZ!>9XS-#C(YX+5Dd!z=5|D63guZ!cnP}r+*rA`Q z42Z{wp+oc~;v*x7AdM#cm%1njOfk!9F?&RXX<&D$qW3F(y=NXv@PiH#CB1U@z^Bli z4sNZV)t~l331T3a6K`|Cg9SlMG)}RP`j7mVFX^g$VbsgjW6u3s1VCT}-5&|as+t*i zVQ&ZsM4_har_ILMv3e=FdXpM2R#gJIKd=ms&?6>8?w8i9>wzng)Y-3d))~~7$@-#` zyWO~s!`x9Zf=BQM#0JK)nB5xe$u$tR;vzJl$Cd^G_8Wg2Udk9j$oN!Owoj@0Cu*g)D|>SC~YLm^)feL6~`(mm#I zEoz_z69LZ?bUX*IJsd`J!#l%UbJ;S2_bL zG+NWbq1+%nP)12?_|ALZ@Gy)UM2z!ez3d_fbF( z@&w3>9`0RRtosv3{ar^iT$vs&pl)PARx(zrQ!*jpS$pa~zpI)fIEVSTfvALPBet2e zl1i`R3z^l?G2Gf&-%O&Zx2x?4n^1mUe{?(xxz0ysCY#mt^$RtHYCqaLwMMt%)a9hc zdaJbFB!TWz-m{85u>~a^@FpBI;Cr2q};4kPk>3Y>%dJoIU`(}MG?5Oxf3hS0$Rm{)0iEv&!-$nMWNmStF-S(or}Y*y5T4+Bf4yf* zzl`~N`y(Cl+*)qV1uQm3=B!njCj7ZGu0y^LSEGSq`TR7qOH#d^@>~x?9&<7>b&STf z>AA$=9}VV(yz=sXrYU~jDFr2tkz-#?4M+e>Tdku6+W?f32G&rbR&fXQebGAUIvS(U zDn>h+11RXje6n|ci>zY8br?Fy6mKl^EOaYklo0N-d5iY=>Jty=66r+A^(vP64fea| z;r}BH2w8MNg}?b;{q{F)1n7VZJ<1y5XN#PPNH?{Elo`j;}U%$EU&a3`tW zF@wOYx1GKecY+Vq2C;L8VkY1j*<)RYE6OgpgW|B#=42@Wj&eiKE*j(6H2i;MwzHm# zmu-{R-CTw_Id%1{#kGdH*DYi?%vt z<%M1nT%PAYm{7b<*{H1gA(4#uEy#vg%nB&G*L?#l5LWBN1dp@aL4+eJ*pj=nd zH+rEYZBNU1M4JObG(l0M0n5H5x#2pB5u*Jdafl0XHIq^Te=S_FfvKG*fN@aY!kate zH1=8~2A+(Z$8V9nJ$Nt#mf^xJbUv&=gCBeB?(}T-lljY`s}sF8J>gfMIC7Hk@YLME zbxq=7H$2aP?SIyIGhAyI)OUo&U_x?|XZ)dwF0hN6H3g_(8w-hHi62Ibjgi{|(|^o+ z_LY|sa|exo@oxT~jIGv8+4ZxA8IhuhdJj~nmYdV^Ld}NYU&WvGeVzw)CFrc6ri?c~ z*uR`VSAW)|K6cY|Wo(!njS~~sq5n2hY`^f}Ak*v1if{u@2oS3ZKccv}_yRRPYw(b4 zAiOCFJdCE+4J`@DBew06=fCLYnCc*0q=^@5C&`Z&JF&*^r}2&_H+i-ORrTA%gzV(X zUf&m2XVGE!OsObNrC6{fFz*VD)puWVzLpLC)=y1iPaEwp_lcqJmMRkHvi^RM#n{#& zN`0x>G-*YIQ1cOBTNqk|nu zfyK9U_WQXHvV3)Fl0e^>V$TL{cRVtut>rt?KF%J4zxq3mhcmnmI0ba|G*WclFLx9r^^Ex;#irEg;5hE9PTcm1A?XjWyg>8JXF(Y z>%ny;7#pwF*}vgx170ypg-KlflEyD{Y#9OL#=AkoYvGe|8pvu~WnYxZVwWEzdl})J zM2@XNGu9pt*wU3!YSz$-&>HQ3+*noQ3%Z*IDdAEK^i_eO^RTm`Va_Kc!WCQ&rBFw_4s67BqJS7i+vwg zP`he=bh57mpere9PBfFvc<{h5^*z3>WQ2!V(#A$z z4mLK2Z-cYZj?Dr!b@cf#i4BLA1Vez?07Nvyfu`)Qp1;}%QzF2mv*5gt&@kGKdq+XV zi5%4}ODA?!>)T2PjUD*LyfLkJrNx?sLUCQY;CZuBVby*FnjkIX2sc3irz20n5^5GD z!=0w|;D9GU;NrGs?;=P7pAW1@?t;~K_`0dO&f7RWGh>SIOpOMsCizGZZD_0khsx^CDH8}s5_7CV(doB>qfLb@1EJ5P zitS4Xd;#TH#l(`E)|iAk3uVZhji`DSwI8y%Ndl2k*J@hUQ~4v zprUYaKcwe;3bJmD3awUN%O4QPk3db;$~aOZ;o~9{|84W8C_R~zr9N?s53k`X$(vZV z5m$-^k#Fs>77#8Gqq(A`D3kHopQb!Ev9GvlBBwG-cUE+GfZAk6?o>sd3CsxJbx>=p z-q{V1A<^})=RN^W`>kmpB{2e5$)el(qZ0;me zSf*qCj#b2juVS(XQ5sPA6#iIG)s=>2l_eTPpk;<_KqvaLayl6yda~2x$ajZ{phd;!v@ktV9Poe5(u|3KJUDcba zDyXW{y&KR``b$_!Uqik39zYdr8uk`JV(Xf#(fXqz!BgZ+&V z;}0}v%_1P4&!`09WNknR4+)cQVg8Qz6&?K3&(2}}a$UR^ta2c;OqM`WX zaT+Y4OYPg1^}{=(x>%dFis9$@*%^GL8go_yUjmTTJy3-teq_em`3*L!nI05tyA?C&hg~~RW}--@!ET$Lx_$w<_)Raiqs>y6-GF){xuJM)QE8aCTfqGjQPeb3J~(?C~>( z(3OZ+72LZy+r8d9nAa6Qkgy@BeQEs|Zv96HZkg|}<^#Js-L{fJg9NL|9##yGE?xh5hHtfCiM@n${&>OxHQq`sYB89hDzuJ)%_U5)|}6!#W;D zqn{T(7T~mJF6f1Vn@rjM8o@Y3xIiVvo=0t>yc(;{Wa+B@Xanm2tN~6R7_+SVBA4L^ z*YJ_KSLqKtKj(+jQYY|U9ao)~2LOHoyxqM0RUuaS=Ot$_2xJn7K><~G)Wz`nCxBG_ zGBglun{m0w4Gl2G#mjKCd7q}A2c2U*B1lqqqJy&`4Z zbxm83kHmt$yGGjnvA$al``mjh@r~&(&jxj#&r2!{*Q5ZmR?-Ru1tVF%!s7bS0x9HOzpyQs z)q)0N#3kUiC&xvdqj_1|d}@9HmkJu`0h+T0yBAdcxYjdX2KfygKEDFIQ+0V;{>R!v z^g5X&Eh!XSn9O?!Q+YbTaD8rG|LBx6-3USt^?h-;_IIYRV%WNRm^Gic#5%8in!hR2 z#QCY<@NT588*f~rw0=Q4G_m3Q=bT!&a2kzh^O3i`GkIbCS7+_sV$^bof6wK!AltMO z#!tzcV+7ULzzgNQbha|VOE|^o5tKi|K$L-`&q1sP zp2;#XBz6R>ni1Mrki)EOmo<-1>>HnLZ-d6}J#1|x6O`{SQLuzh*_VG@%0^4bvYhv~ z@9>cxRf#?q0BlsEIZ|zbc@>B^{g!{!zYptGwK3}f?SX>PP8z!pQgF!<`$}7OXnv%d|T;P{Um0-j`;Ku3aHp_g)a)u%%{DSO4^1NO%7elcu)TDGv4_-~n z;6Z8wS?Q<}ohPbx zPd!7pLl|Pi+PTo|azK=y#?I0ZM5>Gl<_gx)__z~LT2FceabE1+$>Xzj$+I=3`o&|oGjIwRGM%!n@j0v2uwPb-o!;HE77U%2DL|@VMyBdX zoZfDkx`J$Tep(Mzl;aeeqp?5I~SfkfD1C zpc_12zyh%0uy8oxUs6&@sJ&NrtgfMWXJq0O0RF}$8K00-+rwcPn2SsOTKtWwarcyi za~_TVrw#7E_7@62U&t5d``gU4t^LCNVv!*+!`^LdxqjE; z?)x)8^k=_v4Nb(AAb<~6lm3YFTUm55|M(TF3I)jyZ@qC|B6^Do(lN0Ns5lwXdLR6w z=nIH=I;EO+CDO{IZl>fa!mt#6n669Xz^D=${~}7Gc~c$#3A4^DCXntzt4qwxE`)U9 zNOj{0U~%%RhC-evFuDEMPJ3X(FJ&9lyxq?e6R0sHu++;IqQYADHEec}SPh;-{k;dR zPm0J+qZ;8&I#u2=+%z|Ts!rV9!;i^o*UifGUjK-`kqC1>lr30&6ZdgqDivbR9hd+f zl5#A1&j1)+5@-)Ai!RYEHmgL5Wv~MVrM+pv0x|M=v#((XViV*baFFtWnLf5kxf$t0 zbQ-1MV%sAgMps`Xz37p~bd{4c%qYh$-mt;U*$T@Y-#K$`SL1?A#D+d1`6=d{gi#V1 z|6 z?*_NQ@UIeCx;zf+rb1#d0)Ax$$Bf0{Pins&sX&CmwK0j;bt~yS+5hjI!)}r|WaV=B zmdSIMG(8{{Xa+NA0sZwE3Yrvg>QqTj`W1&%bitTU)6Gcfp!&ijQ#0S z^QuYdcw#PW@1FPlM#iMBUloLkQpjMR^E$JrA2F$(Lqk=Vd*U61WC}$iawmE@DAQu# z^I5q+OkyqDMWynUct@-LHDY=}TH`TMV-I?r{#+Hw&4%M=BEP}wJ+0s|C-mZ3x+j3Z z51gdw0uTS@=dBWbbOLW+%jGGP6VOk|vn+8AT;dq2ngo{B9ldh@H z46u$49nuy3MPo!dnUGJ0p{*;DpNLRHyuajOwY1SYOtoeR|ZE7;PAI&s}M~FAt1ez)Yl~k{czpQtp)x`HeXdcmif*%?CMfl6SyH!$R_U zyv{{!%GPbZqD|}Gacf=AMevUEkH=^_@+=%Vl;-?7dT>`gcKEkFHKlnB5f9`+^ zR7G&>Idl&FyhWp`Ysd!hbIfI!dQ&5Kwd8}I0AybWUw6YXVbd*4!xy0km zR|gNshImx?q@I)5NI}+m+SK%v{a{9Cd|TqxWaf+k<+D8okLNQH2X0((vI!9(>~}E) z9gUjE*Q4*4D+S|Z}KqwOEh;(c4pY zuQb5TCSQg3i2F9Q4&&uBT_YmSjcvhUvC4i@@|G#eKjK8q~ZP7z*+-wzRwZm>>_* zKZ)OxNK(pAjSLbVecu-2Rtr{E28Q&YhvhnQI3+yfUXf`h3fr+k#5_ZKCL6<@gg+4G zivGm@O~gxuL$Kt_Fywa=qNLwnr&=p>6une95HnfJ9$d4?GS<7;36EMj+`(t+ZSy=~ zVEm7UrFoWg)xob~k2V0y_0h&z-=(jc0h?~v;(_R4+;dtw9*z^_{=Nd7L<2W`wL{Iq z0hzjzp-KT4(b6my6I#e;`Wf2srF<93(b;?$MN{ZDu4qE&#*Qg&EH?Y2_uNWg88u%+ zTNSFD=%qbzKGH99^n_w*E7Mos3%;+B}Y7=Cj7>Pm-ivybt2Gi*Z{2Ty>r(=A^x zOW8f$KrGs{mFSoYn>+wz5%+CY11?J~e41aR|07i_z6K}X*(v5gbj&X`pQ4SZ&~LzS zpL=??A~F9hHp4KuUxcr9m<2lra-yA72=9m)sFS?sReR4MD0$2w8#rGUR{LxqrktLn zJ?VYIgX!EN$3`HFC`l5%p;oH~J#BWR(L&$bJ=60R{FBi!d+Z!1Y=f1kSJ!3;cd#9; zu;%Zp;z6Oo%>|NT*p8so0+PM;)>2#?d0RU~RLdU(V(rF10+DtR8v2>(ZKH{qhuir~ zGqr`WB+E|hcu-44Z`ba1fg~!(SFf7J@g5Ae;=af>qr}VXrt5wuChw7#OXvx238tx5!H;xnK+Inp72)AF|ap*E@()05$*c!3bkv0_^(g!qOme;6lgH2Q?3 zolRe2CLZY>RY0Y|?Y~kR^N#H|fto_J#%{e)ZjwjVn~!sNId<-uv0W<~3?tPg=F$Bb za!V3~FAE513^tlX*4j-%kRd*4ukkG_U;E*I?WJ^(4y!yZa5)B?k_6AeTdK7B`_hX- zrDFR)OyAd3A6t*>;}#P2OZo-vAM;m+Nq*vv)-(H5q)|cGprM)AV)T_OhJtTI(b)7{ zw?6bNiZ&Z?@6toQ0No`BF|dWRnGii|spHthjT#!?W7;cNiW!w8%MDY%J8C)?{8Lnl zM~&0$OG5ErwtuDfab{neoA?+WqwkeyqKh%f*8Cso4u?A|jUtW>|^?G^?d+Npz*&u3m0rhN2Ep>Gu(c+*}IW zBeJXAUR+R}M(iha(#&$`D7;$wVM9IrtnUKrL+U!Y*n@FSNrawjqc*wpm_pVwyv$i& zcBkdHTv*pqQdg;ErS=~Z8y5G3H>$Q+k7_V*?RvK;V=*wvjQNEKGN&wKc&75MYn>^Z z%fE)xym2+;C8tIK?{#FoF*|goEY{%freQi-AAdt#qVc{ekVV?glqXoOepx?TPv~{m zAWaZJt06;Urw1iY;wwCHXEuNNRk<|ZDp8j##KxjE{>*^&YjEbVwIzLBL^sphl&PiE zct?}Et(?S2Ea`5%=Ar!a0WC3j3JvtwFC&*oF&y$JZ0;)d@j>!h(us`}LdFG)qR3T_^Pu z^yQb&u1K~bl_aUAyrA3McG2qRB%>gm<=`q?&88;+qmucnv|p?9Rpp~9?P5$lEILaF zO+K2k-^t(HA( zK8-zScg|0Kbnel_wCl>l2ANp)=)r!w@d$*&8@s?ko$1>hJkjlUPWBHpnYDoOr4xgJ zY5t`CMBiE7=e>nxR_r62RCZ4@G6}2ZY?hQs7}qI}y1TxasicYS`2IbTvo6S%+eVV; zyN$(K=S1~FTHY&-Z^Uw6ODma_VU|#&Y+@6-oD*;@{(^WeSns^~>WPhugRPXnfYkV<;$XCSk(m1Rm^6dT8Y)pduZI}!F z0$2YyUi49n_#Ly-f`4r!`@rk(EbpjBBumG-FuKs~Dj2$I857hidgYcK6&Kuv*So3j zZWtF+k6G-j@mORZ(8RAJrZ{;wK{=1*dZ0Z=#G8u;Z+g|)RppwdF>CdG9K##%Qw@^e0>=m69o)@plZ}1ygs=o z(U&LoX>G+fu|!(3hE6{iv0qzBx+ChNldL5~7yNQa1=kHdFtrRlh#32wC_Sbxhx13W zBok#K0bH)C9RVx=0nMZawg1El^>26++T{DFfz@@Hf$EqcOf)A_E8ivJRa*E}DnGnk z^6JaZPZ`13l&Msy`?vyQa8m|!0KKD(GYvqEa6n~iyX6fUEQ&2SgDA5yO zx_9?Dz5^%><`b$m`8}OPok&d7N4T*a^S<*usP3ngQ zJ=Z|pSo0!9aS5k@x1CMqP`_0B_8xUXVB9WB@oGV}xOn7k$QU|LHXmBi6!YFFPSYc$ zC!mVRME|30;`Wvs%a|M3GRT3lcBJR=hk%kvHGFfIDw7&F#*&o3C1rEia3#&h@UPV= zicUZ|iR$DxMn8AUr@}v{#T6Gv$Yn@#YRaJHJ7G&MAHCgj#^IMun%O-iX&Y*t&@Vfm zXXuJnO4sBSw|2>8A5#rb+jn3Sc!S~aAx@%+DwcF}^G=~(RO8TcjrL>1nU@vcMM=HA zQ;Su7W&YBo=p}@6z6e+sdDK+C@C4|FF3=0tK9+m?2-=rK+BEN0)ZQwq)`5BZ@nR!* zFMta7#gM9(wr2v!7a=n1tSmmE0X1Ius?+gfZglTo!J>vG>OI`gM)rqn(d3X}E7Am? zw2$O=-6Cyq-gj5()b$M`7ge?+v8C$NFDqL) zq&;C^1YY?>kQ?@pWkgLJFrXSvjrd+xYBgbK_5L$-OCBtq#LyT&>J+{k1yg{NJVb|@ z{4}?I(3)saId-_i((yNZel3iHEK1cbLiR-HN^c^$L}d<>{>osYSG238H@$#gp3DkU zfjuQ@W>O1RNW+}CGy}u@H>%SzHHuFT(mcCiCr-C=3#(6nlkc1lp^EK`E(0w_pS2$7 z*>!HB^-VsSw&aB?jy}`pBTEENE}Rn8z3IHz!?hQ7 zhm02*qu&0jB@kv8kEDuR1NtwE`P{b!KL@RG?BpRX*zrRX>{sevSz?Rik;LIaq~$5P zelNDCiFnt{r8+rGtiS5if81pBfaW$X^>q1E2}4F%2wgSrg+|N+~%NY?7CwL&6K*` z0jb8Y_370gH3R#e0Ibh#slE-N$Ukhvs2FP*>!8**?Rfb-bXwva$_0cSDc%NgykuSj zj|Hdgp3X3}suw%qZ8f!SuvO}3vbfCUAA0~6TjjI8K0p5a`PSPO;5q;s=*o`NAbUqx zAw&Z0)fCyZ{->8UeE6ewx3OPx73lMJv6<4Y^yPA>-c9T4XEP#Sr7C}f_p2a2^KP8s zLmq+_BEN3`oD_A{Ik_t;7pDeUoULqlSaf3STXgYnwMca(I~IZx!}e{9R96|6<1%+4 z2@eOtkp7xdWngm*3lusVZ2Z{@2=99Nqu(>)q3S8 z0DWOBTbftRezZ%kC{+0M8!w3327Zo~bVfTr^Ks7j4DHtZ=QA)bL@7l!r|W`ILxBC{ z7w4`~eZeoL2~1hXC#1$R$QN%SZFU_zWz*xVs3rIHvG|qQZok9h+IPwqzuUiJ{S2`^ z;y-4#Yi^gjrnbEiO&4ylWG;*yx}Z7{a82`;X&d&Y{~H0sLkC}1nXX@1$~D$_cL0Na zmqCgg&qZVpzs^1CcnE1CbCPWd*kXOTQ;DC;7qf&eK44;O!?vuCy7!5YT)7g`0s!UL0BG z=+h?v&%R`_6aWL}*MOemJu1|oLI&6Vu0CpIk*i}cctN2*{yK4Jul7}${~vk6h?kNU z3=I_ghX4C18Sc}fj`oj|`K**%t`;YFk2K1kvFcH=Y^04g;_9sS%VBjSs_XBpolnYIPtw`i~2^L=MEQ|GeNixR9jEvd!_mhc6;dttc0rU?Ywp&kt z@Rvn>tdf0dk4rBImfm=e`~5IulBrFWZ&!kj8=BeHknY|bpUD3GCe$@bk}WBl1jc>RSDcD)z@ zqH>XuCY*Z2##Kw4wCkUqa#dEt_Y9o1i=|lXL%)y03+8Rw5&kigL)S=#FKp;TvdNgN zsPiIEzA1Lfnt;ZekAk@57}0;mF|rx7C_&V!Z5(No z7sbEP*r)GVW6FFRu_O$xJg5X_al7)|Ol0${vYhncM79Aym1GD!}CShr|?a z@Z4l34@k84AAN$xT)@=SZXETf!gjJBwnT5?YUz}QV_5VbiFsg zh%Q+J%iN$PPb(_cU|1jhp(@eY&_PfeAx>JE$%k!94vb;mO$V#hB^WZPZvUoB@(nZK zand35D~nvzEBbE+L}7yhF@m?n*)0NJV6c#o<{Rl%Bm7Sh=$hz%oU-xl2dUQGfUX9~hplay;)u_sy?af&?~|$Cwe>@HbkJez#L|u4 z;>WRk$N|c9As%~v-N)-U8i-L=w+bt{Ex*)tF3B@y#(T}4t{V@oa~F4?@^}o&M+$o` zji8~cYi%?{LuOf=E`Yn36dYD$X6MSwBMkPk(dlHU8)*2DV+-+Ea5j6kNxAH83>IpQ z^4qXo9YN}bgflx%cbb_>ifCPjnJydA{DFMXL zL;|RQ^rBReDkVTd5kr$Apnw#SE=UVaK>?B8i&CUWulL~n?zi4Q_x?I-X3aTk&R)-) zXV2bq9HPP@CZ1MkiMg$K0m;}h7m@Os!o{QRJJe7LXT74Z2PARERoZbW ziVYFfIx32ip{kp7(yn0+`x>ImT7%j8nGo~bv~^)dX1~hu@y#g5VO0vCEn-LaElHj* z+ZIaDvB!@V?{XQk;+;`7-VNz*T?w`5;fi0*h%0|@u7u|tj z_GHK}Xw&^!gF<^n$t5|jAwEPvmbCMVjAt7qx4H!3+M|R%8>Dzmy8Og%@#}QAdC>hs zGauf{HEum))GE^Z(#M{%Rhoky;T)&<408m!lbaJjqiLp}@y4)}?+?mmgDVpqv8^TZ z3&sJh>^&5-X^`kR*M?Uu674m_#igEb0c=$fb@bHCHgx={gj8%h<7m@Waa5kTZW>-2 zd{QW95@@Sy5ixWQ6qyEhiHr;aV<8JPkl-W{qJIq6G!m+CJG>o<;0|tnaZuzn4?P_& z%=Eb|Y8e>Pl0;7WiR2#TDl~2VDlj(yg@0T^!ygZMOYJ*9dWAGRF=c!6AoPSc!kSpE zucpZO424*gyD!6Yuk00w^b;Kj5qlNsj4h9}VTsLDzG>F(68l*$ne=#t_8qDMsb6(K zb6DBP^g`|94pp@Ahgo^#8bs7<9z4K)V*o>cJwtR?3I@B2A|Q8&5)PaS6Br$23z`ZwG5_vHYoOX^n{F{{vlgOysX=A91! zHx|ya%zcES)mxuusm75X4Dm7+dXb~q#<7+QCE=`qDU4rw*#c`I-4U; z2cgr~$}z4pFXvQ72Zo$` zUz_&^)Ex2Rler$mb?NF|Cc&yR5b5X49@P7`<(1L=oozv^rH@3@USotjp!Mut>1WJQ zb}k$z2)b>eyIUO^+gV7}Lvb2!rMzq~#UzyD4=w0LSwt3-tR6}85M=R6Pl6|-woMS(`-(G`b|Av3(#=-HQ<^@>k@w|dY z^*QYoJk&F$K?KBbPx*6B+_ECJHYsA1^R&A08(A^V1`P&HVFWnKb57&}5S&jpwa!4C ziI$~}j!q>0a^PV5s~w!w@1>hjWcmbEAH|SrW2WH>79I>Raeq#=)>X$?IFs`M`SUdv zUN5(cWin)3E)6v#I5OYbVRe{jhc3cdG0 zT!~oo>J60Pjl{Gm;xnB1@VR=VewYi>8rWzZpFmN<_k^;790E^J-Hx8%IYymIq*%K?IWoTCk zZ32D*;x|Jv&NislYvsJ$ic+jmlUlx%jjbCTYg4uZV*yo8F*T=i-d)G{Ke{ed-EG>o zXF@_D$aAsS&+0uxb!*H*yUQz3da(*zvMXk2Q zJ7v3@cx5{Iap%eFv8;3Q0Z?5;er0U^c+kLxUBgA0)FT6G!#bT<>~PDV?51|-#H#^w zxuqvoJbML;eL}vFRg&*VDS77s|tUKa{g@T04<=_Eg zFSFEFJj`DuM|_ppIz7!)h;p;(*ik#Ri`V8O-iKbBx&8jjFm26mfdwLM3{u;d>R#59 zyW~2aflRESsEFrjRL0>Bm-Asa$j97!jEWQAl0XZ~ggP_$Rd>=yOHVkj)L>v%Rn~kC zLJEKAoGmDS%UEU+oq%nC{R&WZOPgjt0bLFEAi+=M#CAufzG}2ACkC|yDWAMMhdxBP1WfxNp*jB0Qy)Y-@y3^oz{^V$5;g-!F3<^_PDE)!jVOicNvV=7!5o) zCpk`ts&Iqq>W!|%p!8&2oVUgSAz6vfsGv=_WGObXI)g_HA}?W&XoEzLz#Qkw!&d;H z@wA6tW`eFxVe~#aGN0ux@$HR7SjIC2s4RT#0CdWwZS`Jy2M!D?Rd%eFgDgwsR>T{Y z!PToSB4Fb3?WBzmT0ZglNv#q%3Nu+?%!GH6!|`EfTwC4%*Q9nfo>6&7OU*X1d;s7YdZ!nsi9!^ za#sw+>oxU!CBA&jvfFs~|A3iyMw`V;kS}~HSF}$R6Q;BX8Q?_edv?
  • $WNFtwhRNO@B=>tx@c`L|a;NU8V)m_%3vn z^DJ9s5eY{&AUS=S6+?XPBcQtz0aXo-g_R4n&Dwg(&&D;nKv65VWRiyt*n_V*TG&0$1&W>!pT>SC65Ywf$1w+yf zCkAQ4c&C#mAK(_0t#CR@cC!t%&;jw%;pfu4{qU|yCC*>BYn>V9a4XgK!W8sG_gAr> zrvCQ?X8yd>#B2Po_ySiQN}*Y9T)}=6=Nd%hW&GxmQ;RSlfl?+!``auy>zvviV!CDFJ82ZV68d#X(l_*;wBv_4?Mcy3n|u zUwxk6+6)R%2{HfnEi@Ur36-J#LH{+BAjl|%ue&}~HzQ0Z2S%(KbtZ3JcMTmu&?dSO z9}l0TOiP`sn04>>;`yh(6Lk*J8^U`8j9HXGrqM%ywE&fVf_)Y2Vt;PS&X|)iJW|cC zz!+<7mR_8Hfvd5P>{asUXR(*Q+S10A^q%P+sP;-kWXio5Hem}UpL5y5p3&LzIuT(D z^ZHgwDE2x@u4AgZTvQzFuhOk3nf;ySkt}*kZrlEQ6RIa$A6WPqBY@?JpSb9H?I$lD z-qypepeA>BITNY;#yY>f))Bs0D+Y5E@ATE)wvR~>cPshiui+B4rx{vf(g)2bUPYhY z$Wy;CWk0X_=k);BLq@Gv4+gmE7>{sq$+!9&!pK?4zpQYGR{9mcR-rUX22B2S`L#T7 zpwEHnPrZ$3j~3zY;LG_Vm&kAxm*k|V|8}-DM5tf92@-RKatJtGVkmsY=vg^*KKhW6 z*U$0`fBchb}SQW;LC zg2;i6lc2=!1S_VkX2~~eKzu)0H4b=cVSSP#r{QA_qTD~`?qkV1!Wt~ulyDZ=)YlFN zOTAZkNaMU{PIh0*zSJZp{tW$TxoMLl)P{NP*yx Date: Wed, 2 Sep 2026 12:04:36 -0700 Subject: [PATCH 4/6] chore(legal): clarify DPF certification scope (#7406) --- apps/sim/app/(landing)/privacy/privacy-content.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/sim/app/(landing)/privacy/privacy-content.tsx b/apps/sim/app/(landing)/privacy/privacy-content.tsx index 1ac09e9df8f..ca6d93d4e9c 100644 --- a/apps/sim/app/(landing)/privacy/privacy-content.tsx +++ b/apps/sim/app/(landing)/privacy/privacy-content.tsx @@ -43,7 +43,7 @@ function richText(content: string): ReactNode { export const PRIVACY_CONFIG: LegalPageConfig = { title: 'Privacy Policy', description: 'Sim Studio, Inc. · Operating the Sim platform (sim.ai)', - lastUpdated: 'August 25, 2026', + lastUpdated: 'September 2, 2026', intro: [ { kind: 'paragraph', @@ -399,7 +399,7 @@ export const PRIVACY_CONFIG: LegalPageConfig = { { kind: 'paragraph', content: richText( - 'Sim subjects all Personal Data received from the European Union, the United Kingdom and Gibraltar, and Switzerland in reliance on the applicable part of the DPF program to the relevant DPF Principles. Sim Studio, Inc. has no other U.S. entities or U.S. subsidiaries covered by its certification. This public policy covers non-human-resources Personal Data. Any human-resources data covered by the certification is addressed in the applicable employee privacy notice.' + "Sim subjects all Personal Data received from the European Union, the United Kingdom and Gibraltar, and Switzerland in reliance on the applicable part of the DPF program to the relevant DPF Principles. Sim Studio, Inc. has no other U.S. entities or U.S. subsidiaries covered by its certification. Sim's certification under the EU-U.S. DPF, the UK Extension to the EU-U.S. DPF, and the Swiss-U.S. DPF covers non-human-resources Personal Data only. Human-resources data is not covered by this certification." ), }, { kind: 'subheading', text: 'Notice, Use, and Choice' }, From abd8f746e7bd6c8a8f6f140f522a427c18c90407 Mon Sep 17 00:00:00 2001 From: Waleed Date: Wed, 2 Sep 2026 12:23:06 -0700 Subject: [PATCH 5/6] fix(slack): disclose paid plan requirement (#7407) --- apps/sim/app/(landing)/integrations/data/landing-content.ts | 2 +- packages/deployment-config/src/integrations.json | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/sim/app/(landing)/integrations/data/landing-content.ts b/apps/sim/app/(landing)/integrations/data/landing-content.ts index eec845c9475..32064764949 100644 --- a/apps/sim/app/(landing)/integrations/data/landing-content.ts +++ b/apps/sim/app/(landing)/integrations/data/landing-content.ts @@ -38,6 +38,6 @@ export const INTEGRATION_LANDING_CONTENT: Record Date: Wed, 2 Sep 2026 12:32:17 -0700 Subject: [PATCH 6/6] fix(deployments): retire inactive-version side effects by row with bounded outbox continuation (#7405) * fix(deployments): retire inactive-version side effects by row with bounded outbox continuation * fix(deployments): fence webhook teardown on version inactivity and abort in undeploy cleanup --- apps/sim/lib/admin/member-operation.test.ts | 5 + apps/sim/lib/admin/member-operation.ts | 4 +- .../billing/enterprise-provisioning.test.ts | 5 + .../lib/billing/enterprise-provisioning.ts | 3 +- apps/sim/lib/core/outbox/service.test.ts | 15 ++ apps/sim/lib/core/outbox/service.ts | 18 +- apps/sim/lib/webhooks/deploy.test.ts | 126 ++++++++- apps/sim/lib/webhooks/deploy.ts | 177 ++++++++++--- .../lib/workflows/deployment-outbox.test.ts | 182 ++++++++++++- apps/sim/lib/workflows/deployment-outbox.ts | 249 ++++++++++-------- .../persistence/deployment-operations.test.ts | 28 ++ .../persistence/deployment-operations.ts | 69 ++++- .../lib/workflows/schedules/deploy.test.ts | 92 ++++++- apps/sim/lib/workflows/schedules/deploy.ts | 75 +++++- apps/sim/lib/workflows/schedules/index.ts | 2 + 15 files changed, 874 insertions(+), 176 deletions(-) diff --git a/apps/sim/lib/admin/member-operation.test.ts b/apps/sim/lib/admin/member-operation.test.ts index 445c82dd07f..e915c4b24b2 100644 --- a/apps/sim/lib/admin/member-operation.test.ts +++ b/apps/sim/lib/admin/member-operation.test.ts @@ -51,6 +51,11 @@ vi.mock('@/lib/workspaces/organization-workspaces', () => ({ ownedAttachableWorkspacesWhere: vi.fn(() => undefined), })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), deferOutboxHandler: (reason: string, _minimum?: number, consumeAttempt = true) => ({ outcome: 'deferred', reason, diff --git a/apps/sim/lib/admin/member-operation.ts b/apps/sim/lib/admin/member-operation.ts index 71eb2981bc8..e68090a08e4 100644 --- a/apps/sim/lib/admin/member-operation.ts +++ b/apps/sim/lib/admin/member-operation.ts @@ -16,7 +16,7 @@ import { import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats' import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils' import { - deferOutboxHandler, + continueOutboxHandler, enqueueOutboxEvent, type OutboxHandler, outboxEventHasSourceOperationId, @@ -688,7 +688,7 @@ export const processAdminMemberOperation: OutboxHandler = async (rawPay } if (nextWorkspaceIndex < payload.request.workspaceIds.length) { - return deferOutboxHandler('Continuing bounded member workspace moves', undefined, false) + return continueOutboxHandler('Continuing bounded member workspace moves') } } diff --git a/apps/sim/lib/billing/enterprise-provisioning.test.ts b/apps/sim/lib/billing/enterprise-provisioning.test.ts index a71b36f8bb5..58afc58ab15 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.test.ts +++ b/apps/sim/lib/billing/enterprise-provisioning.test.ts @@ -61,6 +61,11 @@ vi.mock('@/lib/billing/webhooks/enterprise-reconciliation-lease', () => ({ ), })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), deferOutboxHandler: (reason: string, minimumBackoffMs?: number, consumeAttempt = true) => ({ outcome: 'deferred', reason, diff --git a/apps/sim/lib/billing/enterprise-provisioning.ts b/apps/sim/lib/billing/enterprise-provisioning.ts index 739ca721211..6a5f1152037 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.ts +++ b/apps/sim/lib/billing/enterprise-provisioning.ts @@ -79,6 +79,7 @@ import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterp import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers' import { env } from '@/lib/core/config/env' import { + continueOutboxHandler, deferOutboxHandler, enqueueOutboxEvent, type OutboxEventContext, @@ -3108,7 +3109,7 @@ export const reconcileEnterpriseMembers: OutboxHandler = async (rawPayl if (!nextCursor) return await context.checkpointPayload({ afterUserId: nextCursor }) - return deferOutboxHandler('Continuing bounded Enterprise member reconciliation', undefined, false) + return continueOutboxHandler('Continuing bounded Enterprise member reconciliation') } export const enterpriseIssuanceOutboxHandlers = { diff --git a/apps/sim/lib/core/outbox/service.test.ts b/apps/sim/lib/core/outbox/service.test.ts index dc4fd2f2504..9817363630c 100644 --- a/apps/sim/lib/core/outbox/service.test.ts +++ b/apps/sim/lib/core/outbox/service.test.ts @@ -25,6 +25,7 @@ vi.mock('@sim/utils/id', () => ({ })) import { + continueOutboxHandler, deferOutboxHandler, enqueueOrReschedulePendingOutboxEvent, enqueueOutboxEvent, @@ -387,6 +388,20 @@ describe('processOutboxEvents — handler success and retry', () => { expect(deferredUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null }) }) + it('re-runs a continued handler without consuming its attempt budget', async () => { + const handler = vi.fn(async () => continueOutboxHandler('continuing bounded cleanup')) + queueTableRows(outboxEvent, [makePendingRow({ attempts: 4, maxAttempts: 5 })]) + holdLease() + + const result = await processOutboxEvents({ 'test.event': handler }) + + expect(result.retried).toBe(1) + const continuedUpdate = updateSets().find( + (set) => set.status === 'pending' && 'attempts' in set + ) + expect(continuedUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null }) + }) + it('dead-letters on failure when attempts reaches maxAttempts', async () => { const handler = vi.fn(async () => { throw new Error('permanent failure') diff --git a/apps/sim/lib/core/outbox/service.ts b/apps/sim/lib/core/outbox/service.ts index a36600aaa43..a4ede767836 100644 --- a/apps/sim/lib/core/outbox/service.ts +++ b/apps/sim/lib/core/outbox/service.ts @@ -81,8 +81,9 @@ export interface DeferredOutboxHandlerResult { minimumBackoffMs?: number /** * Defaults to true for an external acknowledgement with a finite retry - * budget. Set false only for an internal dependency whose own outbox row - * independently reaches completed or dead-letter. + * budget. False is reserved for waits on an internal dependency whose own + * outbox row independently reaches completed or dead-letter, and for + * bounded continuation after durable progress (`continueOutboxHandler`). */ consumeAttempt?: boolean } @@ -100,6 +101,19 @@ export function deferOutboxHandler( } } +/** + * Yields after durable progress so the worker re-runs the event without + * spending an attempt. For bounded batches whose remaining work shrinks on + * every run; a run that made no progress must throw or `deferOutboxHandler` + * instead, or the event never reaches a terminal state. + */ +export function continueOutboxHandler( + reason: string, + minimumBackoffMs?: number +): DeferredOutboxHandlerResult { + return deferOutboxHandler(reason, minimumBackoffMs, false) +} + export type OutboxHandler = ( payload: T, context: OutboxEventContext diff --git a/apps/sim/lib/webhooks/deploy.test.ts b/apps/sim/lib/webhooks/deploy.test.ts index 9732e9d0308..77ce7cc1642 100644 --- a/apps/sim/lib/webhooks/deploy.test.ts +++ b/apps/sim/lib/webhooks/deploy.test.ts @@ -1,9 +1,15 @@ /** * @vitest-environment node */ -import { account, credential } from '@sim/db/schema' -import { queueTableRows, resetDbChainMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing' -import { eq } from 'drizzle-orm' +import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db/schema' +import { + dbChainMockFns, + queueTableRows, + resetDbChainMock, + resetEnvFlagsMock, + setEnvFlags, +} from '@sim/testing' +import { eq, ne } from 'drizzle-orm' import { afterAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest' import type { SubBlockConfig } from '@/blocks/types' import type { BlockState } from '@/stores/workflows/workflow/types' @@ -29,6 +35,14 @@ vi.mock('@/lib/webhooks/utils.server', () => ({ vi.mock('@/lib/webhooks/pending-verification', () => ({ PendingWebhookVerificationTracker: vi.fn(), })) +const { mockIsDeploymentVersionActive, mockIsDeploymentVersionProtected } = vi.hoisted(() => ({ + mockIsDeploymentVersionActive: vi.fn(), + mockIsDeploymentVersionProtected: vi.fn(), +})) +vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + isDeploymentVersionActive: mockIsDeploymentVersionActive, + isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtected, +})) const { mockGetSlackBotCredential, @@ -52,9 +66,11 @@ vi.mock('@/lib/webhooks/providers/slack', () => ({ import { buildProviderConfig, + cleanupInactiveDeploymentWebhooks, resolveTriggerCredentialId, resolveWebhookConfigForBlock, } from '@/lib/webhooks/deploy' +import { cleanupExternalWebhook } from '@/lib/webhooks/provider-subscriptions' import { getBlock } from '@/blocks' import { getTrigger } from '@/triggers' @@ -639,3 +655,107 @@ describe('resolveWebhookConfigForBlock — TikTok routing', () => { expect(result?.error.message).toContain('Reconnect') }) }) + +describe('cleanupInactiveDeploymentWebhooks', () => { + const workflow = { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1' } + const input = { + workflowId: 'workflow-1', + workflow, + requestId: 'request-1', + protectedDeploymentVersionId: null, + limit: 5, + } + + function staleWebhookRow(id: string) { + return { + id, + workflowId: 'workflow-1', + deploymentVersionId: 'version-1', + provider: 'github', + providerConfig: {}, + archivedAt: null, + createdAt: new Date('2026-07-14T08:00:00.000Z'), + } + } + + beforeEach(() => { + mockIsDeploymentVersionActive.mockResolvedValue(false) + mockIsDeploymentVersionProtected.mockResolvedValue(false) + }) + + it('retires one bounded batch of stale rows and reports the remainder', async () => { + queueTableRows(webhook, [ + staleWebhookRow('wh-1'), + staleWebhookRow('wh-2'), + staleWebhookRow('wh-3'), + ]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + + await expect(cleanupInactiveDeploymentWebhooks({ ...input, limit: 2 })).resolves.toEqual({ + hasMore: true, + }) + + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(2) + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledWith( + expect.objectContaining({ id: 'wh-1' }), + workflow, + 'request-1', + { throwOnError: true } + ) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2) + }) + + it('reports completion once the batch drains every stale row', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: false }) + + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(1) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + }) + + it('excludes the version the current operation is preparing from the batch', async () => { + queueTableRows(webhook, []) + + await expect( + cleanupInactiveDeploymentWebhooks({ ...input, protectedDeploymentVersionId: 'version-3' }) + ).resolves.toEqual({ hasMore: false }) + + expect(ne).toHaveBeenCalledWith(webhook.deploymentVersionId, 'version-3') + }) + + it('stops before any provider call once the fence reports a change', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + + await expect( + cleanupInactiveDeploymentWebhooks({ ...input, shouldContinue: async () => false }) + ).resolves.toEqual({ hasMore: true }) + + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) + + it('leaves a row alone when its version was re-activated after the batch was selected', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + mockIsDeploymentVersionActive.mockResolvedValue(true) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true }) + + expect(mockIsDeploymentVersionActive).toHaveBeenCalledWith('workflow-1', 'version-1') + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) + + it('leaves a row alone when its version became the current candidate mid-batch', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + mockIsDeploymentVersionProtected.mockResolvedValue(true) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true }) + + expect(mockIsDeploymentVersionProtected).toHaveBeenCalledWith('workflow-1', 'version-1') + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/webhooks/deploy.ts b/apps/sim/lib/webhooks/deploy.ts index 136720edcc1..7543e0a0595 100644 --- a/apps/sim/lib/webhooks/deploy.ts +++ b/apps/sim/lib/webhooks/deploy.ts @@ -3,7 +3,7 @@ import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateShortId } from '@sim/utils/id' -import { and, eq, inArray, isNull, or } from 'drizzle-orm' +import { and, asc, eq, inArray, isNull, ne, or } from 'drizzle-orm' import type { NextRequest } from 'next/server' import { isSlackExtendedScopesEnabled } from '@/lib/core/config/env-flags' import { getProviderIdFromServiceId } from '@/lib/oauth' @@ -33,6 +33,10 @@ import { replaceSlackStreamAuthoringConfig, } from '@/lib/webhooks/slack-stream-config' import { findConflictingWebhookPathOwner } from '@/lib/webhooks/utils.server' +import { + isDeploymentVersionActive, + isDeploymentVersionProtectedByCurrentOperation, +} from '@/lib/workflows/persistence/deployment-operations' import { buildCanonicalIndex, buildSubBlockValues, @@ -1276,39 +1280,15 @@ export async function cleanupWebhooksForWorkflow( if (!skipExternalCleanup) { for (const wh of existingWebhooks) { - if (shouldDeleteWebhook && !(await shouldDeleteWebhook())) { - logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } - - try { - await cleanupExternalWebhook(wh, workflow, requestId, { - throwOnError: strictExternalCleanup, - }) - } catch (cleanupError) { - logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) - if (strictExternalCleanup) throw cleanupError - // Continue with other webhooks even if one fails - } - - const deleted = await deleteWebhookRecordAfterCleanup({ - workflowId, + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, deploymentVersionId, - webhookId: wh.id, + strictExternalCleanup, shouldDeleteWebhook, }) - if (!deleted) { - logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } + if (!deleted) return } } else { for (const wh of existingWebhooks) { @@ -1336,6 +1316,141 @@ export async function cleanupWebhooksForWorkflow( ) } +type WebhookRow = typeof webhook.$inferSelect + +/** + * Tears down one webhook's provider subscription and then deletes its row. + * Returns false when `shouldDeleteWebhook` reports the deployment became + * active again, in which case the caller must stop touching its rows. + */ +async function cleanupWebhookRow(params: { + webhook: WebhookRow + workflow: Record + requestId: string + deploymentVersionId?: string | null + strictExternalCleanup: boolean + shouldDeleteWebhook?: () => Promise +}): Promise { + const { webhook: wh, workflow, requestId, deploymentVersionId, strictExternalCleanup } = params + const workflowId = wh.workflowId + if (params.shouldDeleteWebhook && !(await params.shouldDeleteWebhook())) { + logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + return false + } + + try { + await cleanupExternalWebhook(wh, workflow, requestId, { throwOnError: strictExternalCleanup }) + } catch (cleanupError) { + logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) + if (strictExternalCleanup) throw cleanupError + } + + const deleted = await deleteWebhookRecordAfterCleanup({ + workflowId, + deploymentVersionId, + webhookId: wh.id, + shouldDeleteWebhook: params.shouldDeleteWebhook, + }) + if (!deleted) { + logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + } + return deleted +} + +export interface InactiveDeploymentWebhookCleanupResult { + /** True when rows remain beyond this batch and the caller should run again. */ + hasMore: boolean +} + +/** + * Tears down webhooks still owned by inactive deployment versions of a + * workflow, at most `limit` rows per call. Provider teardown costs one call + * per row, so the work is bounded here and `hasMore` asks the caller to come + * back; every finished row leaves the remaining set smaller, so repeated calls + * converge. `protectedDeploymentVersionId` is the version an in-flight + * operation is preparing, inactive until cutover but live preparation state. + * Each row is re-checked right before its provider call: the version must + * still be inactive and must not have become the current operation's + * candidate, since either can change while the batch runs and the fenced row + * delete that follows cannot undo provider teardown. + */ +export async function cleanupInactiveDeploymentWebhooks(params: { + workflowId: string + workflow: Record + requestId: string + protectedDeploymentVersionId: string | null + limit: number + shouldContinue?: () => Promise +}): Promise { + const { workflowId, workflow, requestId, shouldContinue } = params + const inactiveVersionIds = db + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const staleWebhooks = await db + .select() + .from(webhook) + .where( + and( + eq(webhook.workflowId, workflowId), + isNull(webhook.archivedAt), + inArray(webhook.deploymentVersionId, inactiveVersionIds), + params.protectedDeploymentVersionId + ? ne(webhook.deploymentVersionId, params.protectedDeploymentVersionId) + : undefined + ) + ) + .orderBy(asc(webhook.createdAt)) + .limit(params.limit + 1) + + const batch = staleWebhooks.slice(0, params.limit) + if (batch.length === 0) return { hasMore: false } + + logger.info( + `[${requestId}] Cleaning up ${batch.length} webhook(s) owned by inactive deployments`, + { + workflowId, + webhookIds: batch.map((wh) => wh.id), + } + ) + + for (const wh of batch) { + const deploymentVersionId = wh.deploymentVersionId + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, + deploymentVersionId, + strictExternalCleanup: true, + shouldDeleteWebhook: async () => { + if (shouldContinue && !(await shouldContinue())) return false + if (!deploymentVersionId) return true + if (await isDeploymentVersionActive(workflowId, deploymentVersionId)) return false + return !(await isDeploymentVersionProtectedByCurrentOperation( + workflowId, + deploymentVersionId + )) + }, + }) + if (!deleted) return { hasMore: true } + } + + return { hasMore: staleWebhooks.length > params.limit } +} + /** * Deletes a webhook record unless the deployment became active again. * diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 4483575af30..9cfa87dbce9 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -30,6 +30,10 @@ const { mockRecordAudit, mockEmitWorkflowDeployedEvent, mockCaptureServerEvent, + mockCleanupInactiveDeploymentWebhooks, + mockDeleteInactiveDeploymentSchedules, + mockGetProtectedDeploymentVersionId, + mockIsDeploymentVersionActive, mockTx, } = vi.hoisted(() => ({ mockPrepareWebhooks: vi.fn(), @@ -51,6 +55,10 @@ const { mockRecordAudit: vi.fn(), mockEmitWorkflowDeployedEvent: vi.fn(), mockCaptureServerEvent: vi.fn(), + mockCleanupInactiveDeploymentWebhooks: vi.fn(), + mockDeleteInactiveDeploymentSchedules: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentVersionActive: vi.fn(), mockTx: { select: vi.fn(), update: vi.fn(), execute: vi.fn() }, })) @@ -66,6 +74,11 @@ vi.mock('@sim/audit', () => ({ vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), enqueueOutboxEvent: vi.fn(), processOutboxEventById: vi.fn(), })) @@ -85,6 +98,7 @@ vi.mock('@/lib/mcp/workflow-mcp-sync', () => ({ })) vi.mock('@/lib/webhooks/deploy', () => ({ + cleanupInactiveDeploymentWebhooks: mockCleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow: mockCleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy: vi.fn(), saveTriggerWebhooksForDeploy: vi.fn(), @@ -102,7 +116,9 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ activateDeploymentOperation: mockActivateDeploymentOperation, beginDeploymentOperationActivation: mockBeginDeploymentOperationActivation, getDeploymentOperation: mockGetDeploymentOperation, + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + isDeploymentVersionActive: mockIsDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness: mockMarkDeploymentComponentReadiness, @@ -113,6 +129,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ vi.mock('@/lib/workflows/schedules', () => ({ createSchedulesForDeploy: mockCreateSchedulesForDeploy, + deleteInactiveDeploymentSchedules: mockDeleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow: vi.fn(), })) @@ -219,6 +236,10 @@ describe('versioned deployment preparation outbox', () => { }) mockIsDeploymentOperationCurrent.mockResolvedValue(false) mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(false) + mockIsDeploymentVersionActive.mockResolvedValue(false) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + mockDeleteInactiveDeploymentSchedules.mockResolvedValue({ status: 'deleted', count: 0 }) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValue({ hasMore: false }) }) it('activates only after every preparation component is ready', async () => { @@ -558,6 +579,8 @@ describe('versioned deployment preparation outbox', () => { await expect(handler()(payload(), context(new AbortController(), 3))).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).not.toHaveBeenCalled() + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() expect(mockMarkDeploymentOperationFailed).not.toHaveBeenCalled() expect(mockRecordDeploymentOperationRetry).not.toHaveBeenCalled() }) @@ -589,11 +612,35 @@ describe('versioned deployment preparation outbox', () => { { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - await handler()(payload(), context()) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).toHaveBeenCalledTimes(1) expect(mockRecordAudit).toHaveBeenCalledTimes(1) expect(mockEmitWorkflowDeployedEvent).toHaveBeenCalledTimes(1) + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: { + workflowId: 'workflow-1', + operationId: 'operation-1', + generation: 2, + deploymentVersionId: 'version-2', + statuses: ['active'], + }, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ + workflowId: 'workflow-1', + protectedDeploymentVersionId: null, + limit: 20, + }) + ) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) }) /** @@ -619,29 +666,140 @@ describe('versioned deployment preparation outbox', () => { ) }) - it('keeps v1 cleanup from deleting a candidate owned by the current v2 operation', async () => { + it('continues through the outbox while stale webhooks remain, then checkpoints the cleanup', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + expect(outboxContext.checkpointPayload).not.toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledTimes(2) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledTimes(2) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + }) + + it('stops legacy inactive cleanup as soon as its lease is aborted', async () => { + const controller = new AbortController() + controller.abort() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_INACTIVE_SIDE_EFFECTS + ] + + await expect( + cleanupHandler( + { workflowId: 'workflow-1', activeDeploymentVersionId: 'version-2', userId: 'user-1' }, + context(controller) + ) + ).rejects.toMatchObject({ name: 'AbortError' }) + + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() + }) + + it('retires undeployed side effects by row and shields the candidate owned by the current v2 operation', async () => { queueTableRows(schemaMock.workflow, [ { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - queueTableRows(schemaMock.workflowDeploymentVersion, [{ isActive: false }]) queueTableRows(schemaMock.workflow, [{ isDeployed: true }]) - mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-2') const cleanupHandler = createWorkflowDeploymentOutboxHandlers()[ WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS ] - await cleanupHandler( - { + await expect( + cleanupHandler( + { + workflowId: 'workflow-1', + deploymentVersionIds: ['version-2'], + userId: 'user-1', + requestId: 'request-1', + }, + context() + ) + ).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: undefined, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ workflowId: 'workflow-1', - deploymentVersionIds: ['version-2'], - userId: 'user-1', - requestId: 'request-1', - }, - context() + protectedDeploymentVersionId: 'version-2', + limit: 20, + }) ) - expect(mockCleanupWebhooksForWorkflow).not.toHaveBeenCalled() expect(mockCreateSchedulesForDeploy).not.toHaveBeenCalled() }) + + it('continues undeploy cleanup through the outbox before touching MCP tools', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context()) + ).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + + expect(mockNotifyMcpToolServers).not.toHaveBeenCalled() + }) + + it('lets a timed-out undeploy stop between null-version webhooks', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + const controller = new AbortController() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context(controller)) + ).resolves.toBeUndefined() + + expect(mockCleanupWebhooksForWorkflow).toHaveBeenCalledTimes(1) + const shouldDeleteWebhook = mockCleanupWebhooksForWorkflow.mock + .calls[0][6] as () => Promise + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + await expect(shouldDeleteWebhook()).resolves.toBe(true) + + controller.abort() + await expect(shouldDeleteWebhook()).rejects.toMatchObject({ name: 'AbortError' }) + }) }) diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 62875063496..f3462161392 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -3,10 +3,12 @@ import type { PrincipalActor } from '@sim/auth/principal' import { db, workflowDeploymentVersion, workflow as workflowTable } from '@sim/db' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { and, eq, ne } from 'drizzle-orm' +import { and, eq } from 'drizzle-orm' import { NextRequest } from 'next/server' import { env } from '@/lib/core/config/env' import { + continueOutboxHandler, + type DeferredOutboxHandlerResult, enqueueOutboxEvent, type OutboxEventContext, type OutboxHandler, @@ -24,6 +26,7 @@ import { } from '@/lib/mcp/workflow-mcp-sync' import { captureServerEvent } from '@/lib/posthog/server' import { + cleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy, saveTriggerWebhooksForDeploy, @@ -44,7 +47,9 @@ import { beginDeploymentOperationActivation, type DeploymentOperationGeneration, getDeploymentOperation, + getProtectedDeploymentVersionId, isDeploymentOperationCurrent, + isDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness, markDeploymentOperationFailed, @@ -52,7 +57,11 @@ import { setDeploymentTxTimeouts, type WorkflowDeploymentOperation, } from '@/lib/workflows/persistence/deployment-operations' -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from '@/lib/workflows/schedules' +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from '@/lib/workflows/schedules' import { emitWorkflowDeployedEvent } from '@/lib/workspace-events/emitter' import type { BlockState } from '@/stores/workflows/workflow/types' @@ -77,6 +86,16 @@ export const DEPLOYMENT_READINESS_COMPONENTS = ['webhooks', 'schedules', 'mcp'] */ const DEPLOYMENT_PREPARATION_MAX_ATTEMPTS = 4 +/** + * Webhooks retired per outbox attempt when cleaning up inactive deployment + * versions. Each costs a provider call, so the batch keeps one attempt well + * inside the handler timeout; the handler continues through the outbox while + * rows remain. + */ +const INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE = 20 + +const INACTIVE_CLEANUP_CONTINUATION_REASON = 'Continuing inactive deployment side-effect cleanup' + interface DeploymentPreparationCheckpoints { webhooksPrepared?: boolean schedulesPrepared?: boolean @@ -132,7 +151,12 @@ interface SyncActiveSideEffectsPayload { interface CleanupUndeployedSideEffectsPayload { workflowId: string - deploymentVersionIds: string[] + /** + * Versions the undeploy retired. Cleanup finds stale rows from the versions' + * current state instead; kept for one release so events written by earlier + * pods still parse, and events written here still parse on them. + */ + deploymentVersionIds?: string[] userId: string requestId?: string } @@ -249,7 +273,7 @@ function createPrepareDeploymentHandler( return async (rawPayload, context) => { const payload = parsePrepareDeploymentV2Payload(rawPayload) try { - await prepareDeploymentOperation(payload, context, prepareWebhooks) + return await prepareDeploymentOperation(payload, context, prepareWebhooks) } catch (error) { const isFinalAttempt = context.attempts + 1 >= context.maxAttempts if (isNonRetryableDeploymentError(error) || isFinalAttempt) { @@ -305,7 +329,7 @@ async function prepareDeploymentOperation( payload: PrepareDeploymentV2Payload, context: OutboxEventContext, prepareWebhooks: PrepareDeploymentWebhooksHook -): Promise { +): Promise { context.signal.throwIfAborted() let operation = await getDeploymentOperation(payload) context.signal.throwIfAborted() @@ -338,7 +362,7 @@ async function prepareDeploymentOperation( * whatever else has started since, while only the fenced cleanup is * skipped. */ - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -346,7 +370,6 @@ async function prepareDeploymentOperation( checkpoint, context, }) - return } if (operation.status !== 'preparing' && operation.status !== 'activating') return @@ -488,7 +511,7 @@ async function prepareDeploymentOperation( notifyMcpToolServers(affectedMcpServers) context.signal.throwIfAborted() - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -517,6 +540,10 @@ async function prepareDeploymentOperation( * them on the same predicate would drop them for good in the window where a * newer generation exists but has not activated — this activation is still * the live one there, and nothing else will emit them. + * + * Inactive-version cleanup is bounded per attempt. While rows remain, the + * handler yields a continuation so the outbox re-runs it without spending an + * attempt; the notifications above are checkpointed and never repeat. */ async function runPostActivationWork(params: { payload: PrepareDeploymentV2Payload @@ -525,20 +552,21 @@ async function runPostActivationWork(params: { checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { +}): Promise { await emitPostActivationSideEffects(params) await cleanupRetiredWebhooksForOperation({ payload: params.payload, workflow: params.workflow, context: params.context, }) - await cleanupInactiveDeploymentsForOperation({ + const cleanupComplete = await cleanupInactiveDeploymentsForOperation({ payload: params.payload, workflow: params.workflow, checkpoints: params.checkpoints, checkpoint: params.checkpoint, context: params.context, }) + return cleanupComplete ? undefined : continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } async function prepareReadinessComponent(params: { @@ -622,14 +650,19 @@ async function cleanupRetiredWebhooksForOperation(params: { }) } +/** + * Returns false while inactive-version cleanup still has rows to retire, so + * the caller yields a continuation instead of completing the event. A + * superseded attempt returns true: the newer generation owns the cleanup now. + */ async function cleanupInactiveDeploymentsForOperation(params: { payload: PrepareDeploymentV2Payload workflow: Record checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { - if (params.checkpoints.inactiveCleanupCompleted) return +}): Promise { + if (params.checkpoints.inactiveCleanupCompleted) return true const operationFence = { workflowId: params.payload.workflowId, operationId: params.payload.operationId, @@ -644,18 +677,18 @@ async function cleanupInactiveDeploymentsForOperation(params: { return isCurrent } - if (!(await shouldContinue())) return - await cleanupInactiveDeploymentVersions({ + if (!(await shouldContinue())) return true + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: params.payload.workflowId, - activeDeploymentVersionId: params.payload.deploymentVersionId, workflow: params.workflow, - userId: params.payload.userId, requestId: params.payload.requestId, shouldContinue, operationFence, }) - if (!(await shouldContinue())) return + if (!(await shouldContinue())) return true + if (!complete) return false await params.checkpoint({ inactiveCleanupCompleted: true }) + return true } async function emitPostActivationSideEffects(params: { @@ -898,9 +931,10 @@ const syncActiveSideEffects = async (rawPayload: unknown): Promise => { }) } -const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => { +const cleanupInactiveSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupInactiveSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -909,18 +943,19 @@ const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => if (!workflowRecord) return - await cleanupInactiveDeploymentVersions({ + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: payload.workflowId, - activeDeploymentVersionId: payload.activeDeploymentVersionId, workflow: workflowRecord as Record, - userId: payload.userId, requestId, + shouldContinue: unlessAborted(context.signal), }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } -const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise => { +const cleanupUndeployedSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupUndeployedSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -930,47 +965,45 @@ const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise if (!workflowRecord) return const workflowData = workflowRecord as Record - for (const deploymentVersionId of payload.deploymentVersionIds) { - const [versionRow] = await db - .select({ isActive: workflowDeploymentVersion.isActive }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, payload.workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId) - ) - ) - .limit(1) - - if (!versionRow || versionRow.isActive) continue - await cleanupDeploymentVersionIfInactive({ - workflowId: payload.workflowId, - workflow: workflowData, - userId: payload.userId, - requestId, - deploymentVersionId, - }) - } + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: payload.workflowId, + workflow: workflowData, + requestId, + shouldContinue: unlessAborted(context.signal), + }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) + context.signal.throwIfAborted() await cleanupNullVersionWebhooksIfStillUndeployed({ workflowId: payload.workflowId, workflow: workflowData, requestId, + signal: context.signal, }) + context.signal.throwIfAborted() await removeMcpToolsIfStillUndeployed(payload.workflowId, requestId) } +/** Continuation gate for handlers without an operation fence: stops only when the lease aborts. */ +function unlessAborted(signal: AbortSignal): () => Promise { + return async () => { + signal.throwIfAborted() + return true + } +} + /** * Run inactive-version cleanup synchronously as part of the active-version sync, right * after the active version's webhooks/schedules are registered. * - * {@link cleanupInactiveDeploymentVersions} re-checks that each version is still inactive - * before tearing anything down, so it can never touch the now-active version. Running it - * inline — rather than only enqueueing it — closes the window where a lost + * {@link cleanupInactiveDeploymentSideEffects} only selects rows whose version is inactive and + * re-checks each webhook right before its delete, so it can never touch the now-active version. + * Running it inline — rather than only enqueueing it — closes the window where a lost * `CLEANUP_INACTIVE` outbox event leaves superseded webhooks behind as live-but-never-polled - * `is_active` orphans. The deferred event is kept as a fallback so cleanup still retries if - * the inline pass throws, without failing the already-succeeded registration. + * `is_active` orphans. The deferred event is kept as a fallback so cleanup still continues if + * the inline pass throws or has more rows than one bounded pass retires, without failing the + * already-succeeded registration. */ async function syncInactiveDeploymentCleanup(params: { workflowId: string @@ -980,57 +1013,64 @@ async function syncInactiveDeploymentCleanup(params: { requestId: string }): Promise { try { - await cleanupInactiveDeploymentVersions(params) + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + }) + if (complete) return + logger.info( + `[${params.requestId}] Inline inactive-deployment cleanup has more rows; continuing through the outbox` + ) } catch (cleanupError) { logger.warn( `[${params.requestId}] Inline inactive-deployment cleanup failed; deferring to outbox retry`, cleanupError ) - await enqueueWorkflowInactiveDeploymentCleanup(db, { - workflowId: params.workflowId, - activeDeploymentVersionId: params.activeDeploymentVersionId, - userId: params.userId, - requestId: params.requestId, - }) } + await enqueueWorkflowInactiveDeploymentCleanup(db, { + workflowId: params.workflowId, + activeDeploymentVersionId: params.activeDeploymentVersionId, + userId: params.userId, + requestId: params.requestId, + }) } -async function cleanupInactiveDeploymentVersions(params: { +/** + * Retires schedules and webhooks still owned by inactive deployment versions + * of the workflow. Work is keyed by side-effect rows, never by versions, so a + * workflow deployed hundreds of times costs no more than one deployed twice. + * Schedules go in one fenced statement; webhooks need a provider call each + * and drain in bounded batches, with `complete: false` asking the caller to + * run again. `shouldContinue` gates every step and throws once the outbox + * lease is aborted. + */ +async function cleanupInactiveDeploymentSideEffects(params: { workflowId: string - activeDeploymentVersionId: string workflow: Record - userId: string requestId: string shouldContinue?: () => Promise operationFence?: DeploymentCleanupOperationFence -}): Promise { - if (params.shouldContinue && !(await params.shouldContinue())) return - const inactiveVersions = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, params.workflowId), - ne(workflowDeploymentVersion.id, params.activeDeploymentVersionId), - eq(workflowDeploymentVersion.isActive, false) - ) - ) +}): Promise<{ complete: boolean }> { + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } - for (const version of inactiveVersions) { - if (params.shouldContinue && !(await params.shouldContinue())) return - if (await isDeploymentVersionProtectedByCurrentOperation(params.workflowId, version.id)) { - continue - } - await cleanupDeploymentVersionIfInactive({ - workflowId: params.workflowId, - workflow: params.workflow, - userId: params.userId, - requestId: params.requestId, - deploymentVersionId: version.id, - shouldContinue: params.shouldContinue, - operationFence: params.operationFence, - }) - } + const schedules = await deleteInactiveDeploymentSchedules({ + workflowId: params.workflowId, + operationFence: params.operationFence, + }) + if (schedules.status === 'superseded') return { complete: false } + + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId(params.workflowId) + const { hasMore } = await cleanupInactiveDeploymentWebhooks({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + protectedDeploymentVersionId, + limit: INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE, + shouldContinue: params.shouldContinue, + }) + return { complete: !hasMore } } async function cleanupDeploymentVersionIfInactive(params: { @@ -1155,25 +1195,6 @@ async function cleanupStaleDeploymentIfNeeded(params: { return false } -async function isDeploymentVersionActive( - workflowId: string, - deploymentVersionId: string -): Promise { - const [versionRow] = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId), - eq(workflowDeploymentVersion.isActive, true) - ) - ) - .limit(1) - - return Boolean(versionRow) -} - async function removeMcpToolsIfStillUndeployed( workflowId: string, requestId: string @@ -1194,12 +1215,18 @@ async function removeMcpToolsIfStillUndeployed( notifyMcpToolServers(tools) } +/** + * The per-row gate also throws once the outbox lease aborts, so a timed-out + * undeploy stops between webhooks instead of overlapping its reaped retry. + */ async function cleanupNullVersionWebhooksIfStillUndeployed(params: { workflowId: string workflow: Record requestId: string + signal: AbortSignal }): Promise { const isStillUndeployed = async () => { + params.signal.throwIfAborted() const [workflowRecord] = await db .select({ isDeployed: workflowTable.isDeployed }) .from(workflowTable) @@ -1458,7 +1485,7 @@ function parseCleanupUndeployedSideEffectsPayload( const record = parsePayloadRecord(payload) const workflowId = parseRequiredString(record.workflowId, 'workflowId') const userId = parseRequiredString(record.userId, 'userId') - const deploymentVersionIds = parseRequiredStringArray( + const deploymentVersionIds = parseOptionalStringArray( record.deploymentVersionIds, 'deploymentVersionIds' ) @@ -1467,7 +1494,12 @@ function parseCleanupUndeployedSideEffectsPayload( ? record.requestId : undefined - return { workflowId, deploymentVersionIds, userId, requestId } + return { + workflowId, + ...(deploymentVersionIds ? { deploymentVersionIds } : {}), + userId, + requestId, + } } function parseCleanupInactiveSideEffectsPayload( @@ -1509,12 +1541,13 @@ function parseRequiredPositiveInteger(value: unknown, fieldName: string): number return value } -function parseRequiredStringArray(value: unknown, fieldName: string): string[] { +function parseOptionalStringArray(value: unknown, fieldName: string): string[] | undefined { + if (value === undefined) return undefined if ( !Array.isArray(value) || value.some((item) => typeof item !== 'string' || item.length === 0) ) { - throw new Error(`Deployment outbox payload is missing ${fieldName}`) + throw new Error(`Deployment outbox payload has an invalid ${fieldName}`) } return value } diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts index 757e54eb2e8..1a0ba58f210 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts @@ -21,6 +21,7 @@ vi.mock('@sim/utils/id', () => ({ import { activateDeploymentOperation, + getProtectedDeploymentVersionId, markDeploymentComponentReadiness, markDeploymentOperationFailed, prepareWorkflowDeployment, @@ -489,3 +490,30 @@ describe('deployment operation persistence', () => { expect(dbChainMockFns.update).not.toHaveBeenCalledWith(schemaMock.workflow) }) }) + +describe('getProtectedDeploymentVersionId', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('returns the version the in-flight current operation is preparing', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'preparing' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBe('version-3') + }) + + it('protects nothing once the latest operation is terminal', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'active' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) + + it('protects nothing for a workflow without operations', async () => { + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) +}) diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.ts b/apps/sim/lib/workflows/persistence/deployment-operations.ts index 9146f35b601..3d5a95517a6 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.ts @@ -70,6 +70,15 @@ export interface DeploymentOperationGeneration { generation: number } +/** + * Identifies the operation a fenced step belongs to, optionally narrowed to + * the version it targets and the statuses it may currently hold. + */ +export type DeploymentOperationFence = DeploymentOperationGeneration & { + deploymentVersionId?: string + statuses?: readonly DeploymentOperationStatus[] +} + export interface WorkflowDeploymentStatus { activeDeployment: { deploymentVersionId: string @@ -282,10 +291,7 @@ export async function getDeploymentOperation( * Confirms an operation still owns the workflow's latest generation. */ export async function isDeploymentOperationCurrent( - params: DeploymentOperationGeneration & { - deploymentVersionId?: string - statuses?: readonly DeploymentOperationStatus[] - }, + params: DeploymentOperationFence, executor: Pick = db ): Promise { const [latestOperation] = await executor @@ -322,6 +328,19 @@ export async function isDeploymentVersionProtectedByCurrentOperation( deploymentVersionId: string, executor: Pick = db ): Promise { + return (await getProtectedDeploymentVersionId(workflowId, executor)) === deploymentVersionId +} + +/** + * The deployment version the current operation is still preparing, or null + * once the latest operation is terminal. Cleanup must leave this version + * alone: it is inactive until cutover, yet its schedules and webhook + * candidates are live preparation state. + */ +export async function getProtectedDeploymentVersionId( + workflowId: string, + executor: Pick = db +): Promise { const [latestOperation] = await executor .select({ deploymentVersionId: workflowDeploymentOperation.deploymentVersionId, @@ -333,12 +352,42 @@ export async function isDeploymentVersionProtectedByCurrentOperation( .orderBy(desc(workflowDeploymentOperation.generation)) .limit(1) - return ( - latestOperation?.deploymentVersionId === deploymentVersionId && - latestOperation.protocolVersion === DEPLOYMENT_OPERATION_PROTOCOL_VERSION && - isDeploymentOperationStatus(latestOperation.status) && - IN_FLIGHT_STATUSES.includes(latestOperation.status) - ) + if ( + !latestOperation || + latestOperation.protocolVersion !== DEPLOYMENT_OPERATION_PROTOCOL_VERSION || + !isDeploymentOperationStatus(latestOperation.status) || + !IN_FLIGHT_STATUSES.includes(latestOperation.status) + ) { + return null + } + return latestOperation.deploymentVersionId +} + +/** + * True when the given deployment version is the workflow's active one. + * Cleanup re-checks this immediately before any provider teardown because a + * version can be re-activated between a batch being selected and its rows + * being processed, and the fenced row delete that follows cannot undo a + * provider call. + */ +export async function isDeploymentVersionActive( + workflowId: string, + deploymentVersionId: string, + executor: Pick = db +): Promise { + const [versionRow] = await executor + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.id, deploymentVersionId), + eq(workflowDeploymentVersion.isActive, true) + ) + ) + .limit(1) + + return Boolean(versionRow) } /** diff --git a/apps/sim/lib/workflows/schedules/deploy.test.ts b/apps/sim/lib/workflows/schedules/deploy.test.ts index afb5dc2abdc..62421cc8fdf 100644 --- a/apps/sim/lib/workflows/schedules/deploy.test.ts +++ b/apps/sim/lib/workflows/schedules/deploy.test.ts @@ -3,12 +3,21 @@ * * @vitest-environment node */ -import { dbChainMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing' +import { + dbChainMock, + dbChainMockFns, + flattenMockConditions, + resetDbChainMock, + schemaMock, +} from '@sim/testing' import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockRandomUUID } = vi.hoisted(() => ({ - mockRandomUUID: vi.fn(), -})) +const { mockRandomUUID, mockGetProtectedDeploymentVersionId, mockIsDeploymentOperationCurrent } = + vi.hoisted(() => ({ + mockRandomUUID: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentOperationCurrent: vi.fn(), + })) vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) @@ -16,7 +25,17 @@ vi.mock('@/lib/webhooks/deploy', () => ({ cleanupWebhooksForWorkflow: vi.fn().mockResolvedValue(undefined), })) -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from './deploy' +vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, + isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + setDeploymentTxTimeouts: vi.fn(), +})) + +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from './deploy' import type { BlockState } from './utils' import * as scheduleUtils from './utils' import { findScheduleBlocks, validateScheduleBlock, validateWorkflowSchedules } from './validation' @@ -795,3 +814,66 @@ describe('Schedule Deploy Utilities', () => { }) }) }) + +describe('deleteInactiveDeploymentSchedules', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + }) + + it('deletes every schedule owned by an inactive version in one statement', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1' }, { id: 'schedule-2' }]) + + await expect(deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' })).resolves.toEqual({ + status: 'deleted', + count: 2, + }) + + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'inArray', + column: schemaMock.workflowSchedule.deploymentVersionId, + }), + expect.objectContaining({ type: 'isNull', column: schemaMock.workflowSchedule.archivedAt }), + ]) + ) + expect(conditions).not.toEqual( + expect.arrayContaining([expect.objectContaining({ type: 'ne' })]) + ) + }) + + it('shields the version an in-flight operation is preparing', async () => { + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-3') + + await deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' }) + + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'ne', + left: schemaMock.workflowSchedule.deploymentVersionId, + right: 'version-3', + }), + ]) + ) + }) + + it('deletes nothing once a newer operation owns the workflow', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(false) + + await expect( + deleteInactiveDeploymentSchedules({ + workflowId: 'workflow-1', + operationFence: { workflowId: 'workflow-1', operationId: 'operation-1', generation: 2 }, + }) + ).resolves.toEqual({ status: 'superseded' }) + + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/workflows/schedules/deploy.ts b/apps/sim/lib/workflows/schedules/deploy.ts index 1f2a1bfd662..475212ffd23 100644 --- a/apps/sim/lib/workflows/schedules/deploy.ts +++ b/apps/sim/lib/workflows/schedules/deploy.ts @@ -1,9 +1,15 @@ -import { db, workflowSchedule } from '@sim/db' +import { db, workflow, workflowDeploymentVersion, workflowSchedule } from '@sim/db' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { and, eq, inArray, isNull } from 'drizzle-orm' +import { and, eq, inArray, isNull, ne } from 'drizzle-orm' import type { DbOrTx } from '@/lib/db/types' +import { + type DeploymentOperationFence, + getProtectedDeploymentVersionId, + isDeploymentOperationCurrent, + setDeploymentTxTimeouts, +} from '@/lib/workflows/persistence/deployment-operations' import type { BlockState } from '@/lib/workflows/schedules/utils' import { findScheduleBlocks, validateScheduleBlock } from '@/lib/workflows/schedules/validation' @@ -203,3 +209,68 @@ export async function deleteSchedulesForWorkflow( : `Deleted all schedules for workflow ${workflowId}` ) } + +export type InactiveDeploymentScheduleCleanupResult = + | { status: 'deleted'; count: number } + | { status: 'superseded' } + +/** + * Deletes every schedule still owned by an inactive deployment version of the + * workflow in one fenced statement. Keyed by schedule rows rather than by + * versions, so the cost follows what is stale instead of how many times the + * workflow has been deployed. The version an in-flight operation is preparing + * is left alone: it is inactive until cutover, but its schedules are live + * preparation state. The workflow row lock serializes this with activation so + * `isActive` cannot flip underneath the delete. + */ +export async function deleteInactiveDeploymentSchedules(params: { + workflowId: string + /** When set, nothing is deleted once a newer operation has taken over the workflow. */ + operationFence?: DeploymentOperationFence +}): Promise { + return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) + await tx + .select({ id: workflow.id }) + .from(workflow) + .where(eq(workflow.id, params.workflowId)) + .for('update') + if (params.operationFence && !(await isDeploymentOperationCurrent(params.operationFence, tx))) { + return { status: 'superseded' } + } + + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId( + params.workflowId, + tx + ) + const inactiveVersionIds = tx + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, params.workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const deleted = await tx + .delete(workflowSchedule) + .where( + and( + eq(workflowSchedule.workflowId, params.workflowId), + isNull(workflowSchedule.archivedAt), + inArray(workflowSchedule.deploymentVersionId, inactiveVersionIds), + protectedDeploymentVersionId + ? ne(workflowSchedule.deploymentVersionId, protectedDeploymentVersionId) + : undefined + ) + ) + .returning({ id: workflowSchedule.id }) + + if (deleted.length > 0) { + logger.info( + `Deleted ${deleted.length} schedule(s) owned by inactive deployments of workflow ${params.workflowId}` + ) + } + return { status: 'deleted', count: deleted.length } + }) +} diff --git a/apps/sim/lib/workflows/schedules/index.ts b/apps/sim/lib/workflows/schedules/index.ts index 67dd11fd2c8..026d35ecd8d 100644 --- a/apps/sim/lib/workflows/schedules/index.ts +++ b/apps/sim/lib/workflows/schedules/index.ts @@ -1,5 +1,7 @@ export { createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow, + type InactiveDeploymentScheduleCleanupResult, } from './deploy' export { validateWorkflowSchedules } from './validation'