ServiceNow Now Assist Cost Control: Business Rule Recursion, Flow Designer Fan-Out, Cross-Table Cascade, and Scheduled Job Overlap

ServiceNow Now Assist embeds generative AI directly into the ServiceNow platform — ITSM incident summarization, HR case analysis, CSM customer interaction drafts, and creator-built AI steps inside Flow Designer workflows. Unlike standalone LLM API calls where every invocation is explicit and logged in your application code, Now Assist invocations are embedded in platform mechanisms that execute invisibly: business rules that fire on database writes, Flow Designer actions triggered by table events, and scheduled jobs that process record queues. The billing model charges per generative AI invocation. Each Now Assist Skill call, each "Generate Now Assist Text" Flow Designer action, each AI-assisted field population — each is a billable event against your organization's Now Assist credit allocation or your per-invocation licensing tier.

Four structural properties of the ServiceNow platform amplify Now Assist credit consumption beyond what's visible when configuring an AI-enabled workflow in the Now Platform designer:

• Business rule update recursion — ServiceNow business rules fire on GlideRecord writes without distinguishing which field triggered the write. A business rule that calls Now Assist to generate an AI summary and writes it back to the same record re-fires itself, because the AI field update is still a record update on the watched table. Recursion runs until ServiceNow's platform-level loop detection (typically 30 nested business rule executions on the same record) trips — burning one Now Assist credit per level.
• Flow Designer ForEach fan-out — Flow Designer's ForEach action iterates over a GlideRecord query result set. If the "Look Up Records" action before a ForEach uses no filter condition or a broad filter (all incidents in "In Progress" state, all open changes, all active tasks), the result set can be thousands of records. Each ForEach iteration that calls a "Generate Now Assist Text" action or invokes a Now Assist Skill consumes one credit per record. A query returning 3,000 incidents × 1 credit per Now Assist action = 3,000 credits from a single flow execution triggered by one event.
• Cross-table trigger cascade — ServiceNow's event system fires flow triggers across table relationships. A Flow Designer flow triggered on Incident update calls Now Assist to generate a problem root-cause draft and writes it to a linked Problem record. The Problem table update triggers a second flow that calls Now Assist to re-analyze the updated problem and writes its findings back to the originating Incident. The Incident update re-fires the first flow. The loop spans two flows and two tables, making it invisible in either flow's execution history when viewed in isolation.
• Scheduled job concurrency overlap — ServiceNow's Scheduled Script Execution (scheduled jobs) can run concurrent instances when a previous run has not completed within the trigger interval. A nightly batch job that enriches all open incidents with Now Assist AI summaries takes 90 minutes to process 5,000 records. Configured to run hourly, two instances overlap: both query the same set of unprocessed incidents, both call Now Assist on each record, and both write back without coordination. Credit consumption doubles, and the duplicate AI writes trigger business rules on every updated record.

Failure Mode 1 — Business Rule Update Recursion

ServiceNow business rules are server-side scripts that execute in response to GlideRecord operations. An "after" business rule configured to run when an Incident record is updated will fire any time any field on that record changes — including fields written by other automated processes, other business rules, and now, by Now Assist itself.

The recursion failure mode is simple to trigger by accident: a developer adds an "after" business rule on the Incident table, condition set to "is update," that calls a Now Assist Skill to generate a plain-language summary of the incident's current state, then writes that summary to the u_ai_summary field using current.u_ai_summary.setValue(summary) followed by current.update(). The current.update() call inside the business rule is itself a GlideRecord update on the Incident table, which immediately satisfies the trigger condition — "is update" on this record — and fires the business rule again. The second invocation calls Now Assist again, writes the summary again, calls current.update() again, and fires the business rule a third time.

ServiceNow's platform-level recursive business rule protection typically terminates the chain after 30 nested executions on the same record within a single transaction. At one Now Assist credit per invocation, a single legitimate Incident update triggers 30 unintended Now Assist invocations before the platform stops the loop. In a high-volume ITSM environment — an incident storm during a P1 outage — hundreds of incidents updating simultaneously means thousands of recursive Now Assist calls from a single event.

The idempotency guard runs as an external HTTP endpoint called from within the business rule script. The business rule calls Now Assist to generate the output, then calls the guard before writing back. The guard returns a should_write decision based on whether the output has changed since the last write:

import time
import hashlib
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "sn_business_rule_guard.db"

HASH_TTL_SECONDS = 3600

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS record_output_hashes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                table_name TEXT NOT NULL,
                sys_id TEXT NOT NULL,
                field_name TEXT NOT NULL,
                output_hash TEXT NOT NULL,
                recorded_at REAL,
                recursive_blocks INTEGER DEFAULT 0
            )
        """)
        conn.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_table_sys_field "
            "ON record_output_hashes (table_name, sys_id, field_name)"
        )

class BusinessRuleRecursionGuard:
    """
    Call check() inside a business rule, after Now Assist generates output
    but before current.update() is called. Returns should_write=false when
    the output hash matches the previously stored hash, preventing the update
    that would re-trigger the same business rule.
    """

    @staticmethod
    def check(table_name: str, sys_id: str, field_name: str,
              ai_output: str) -> dict:
        output_hash = hashlib.sha256(ai_output.encode()).hexdigest()
        now = time.time()
        expiry = now - HASH_TTL_SECONDS

        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute(
                    "DELETE FROM record_output_hashes WHERE recorded_at < ?",
                    (expiry,)
                )
                existing = conn.execute(
                    "SELECT output_hash, recorded_at, recursive_blocks "
                    "FROM record_output_hashes "
                    "WHERE table_name = ? AND sys_id = ? AND field_name = ?",
                    (table_name, sys_id, field_name)
                ).fetchone()

                if existing and existing[0] == output_hash:
                    conn.execute(
                        "UPDATE record_output_hashes "
                        "SET recursive_blocks = recursive_blocks + 1 "
                        "WHERE table_name = ? AND sys_id = ? AND field_name = ?",
                        (table_name, sys_id, field_name)
                    )
                    return {
                        "should_write": False,
                        "reason": "output_unchanged",
                        "table_name": table_name,
                        "sys_id": sys_id,
                        "field_name": field_name,
                        "seconds_since_last_write": int(now - existing[1]),
                        "recursive_blocks_total": existing[2] + 1,
                        "message": (
                            f"Now Assist output for {table_name}.{field_name} "
                            f"on record {sys_id} is identical to the previous "
                            f"write ({int(now - existing[1])}s ago). "
                            "Skipping current.update() to prevent business rule "
                            "re-trigger. This guard has blocked "
                            f"{existing[2] + 1} recursive invocations on this record."
                        ),
                    }

                if existing:
                    conn.execute(
                        "UPDATE record_output_hashes "
                        "SET output_hash = ?, recorded_at = ? "
                        "WHERE table_name = ? AND sys_id = ? AND field_name = ?",
                        (output_hash, now, table_name, sys_id, field_name)
                    )
                else:
                    conn.execute(
                        "INSERT INTO record_output_hashes "
                        "(table_name, sys_id, field_name, output_hash, recorded_at) "
                        "VALUES (?, ?, ?, ?, ?)",
                        (table_name, sys_id, field_name, output_hash, now)
                    )
                return {
                    "should_write": True,
                    "table_name": table_name,
                    "sys_id": sys_id,
                    "field_name": field_name,
                    "output_hash": output_hash,
                }

@app.route("/guard/business-rule", methods=["POST"])
def business_rule_guard():
    data = request.get_json(force=True)
    result = BusinessRuleRecursionGuard.check(
        table_name=data.get("table_name", ""),
        sys_id=data.get("sys_id", ""),
        field_name=data.get("field_name", ""),
        ai_output=data.get("ai_output", ""),
    )
    return jsonify(result), 200

if __name__ == "__main__":
    init_db()
    app.run(port=8096)

In your ServiceNow business rule script, call this endpoint via new sn_ws.RESTMessageV2() after the Now Assist Skill returns output and before the current.update() line. Pass the current table name (current.getTableName()), the record's sys_id (current.getUniqueValue()), the field name being written, and the AI-generated text. Parse the response JSON: if should_write is false, do not call current.u_ai_summary.setValue() and do not call current.update(). The business rule exits without a database write, and no follow-on trigger fires. The recursive_blocks_total counter in the response is useful for monitoring: a persistently high counter on a specific record indicates the business rule is firing repeatedly on that record due to other automation, and the guard is correctly blocking redundant Now Assist invocations each time.

Failure Mode 2 — Flow Designer ForEach Fan-Out

Flow Designer is ServiceNow's low-code automation builder that replaces legacy Workflow for most automation tasks. A flow is triggered by an event (record created, record updated, inbound email, scheduled trigger) and executes a sequence of actions — "Look Up Records," "For Each," "Generate Now Assist Text," "Update Record." The ForEach action iterates over the list returned by a preceding "Look Up Records" action and executes the loop body once per record.

The fan-out failure mode emerges when the "Look Up Records" action uses an insufficiently selective filter. A flow designed to "summarize all incidents assigned to a newly onboarded agent" might use a "Look Up Records" on the Incident table with condition assigned_to is [new_agent_sys_id]. If the newly onboarded agent is receiving incidents from a large backlog migration — 2,000 pre-existing incidents reassigned in one bulk operation — the "Look Up Records" returns all 2,000 at once. The ForEach loop calls "Generate Now Assist Text" for each of the 2,000 records. The flow trigger is one event (agent record updated). The Now Assist credit consumption is 2,000.

The scaling property that makes this dangerous is that ForEach in Flow Designer is not bound by a default iteration limit visible in the designer interface. There is no visual indicator of the expected record count before the flow runs — the designer shows the action configuration, not the runtime query result size. A flow that works correctly when processing 10 incidents per day becomes an uncontrolled credit burn on the day a database administrator runs a bulk reassignment or import job that triggers the same flow trigger condition for 5,000 records simultaneously.

import time
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "sn_flow_foreach_guard.db"

MAX_RECORDS_PER_RUN = 50
MAX_FLOW_EXECUTIONS_PER_HOUR = 10
NOW_ASSIST_CREDITS_PER_FOREACH_ITEM = 1

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS flow_run_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                flow_sys_id TEXT NOT NULL,
                flow_name TEXT NOT NULL,
                execution_id TEXT,
                record_count INTEGER DEFAULT 0,
                credits_approved INTEGER DEFAULT 0,
                started_at REAL
            )
        """)
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_flow_time "
            "ON flow_run_log (flow_sys_id, started_at)"
        )

class FlowForEachGuard:
    """
    Call check() after Look Up Records returns its count but before ForEach begins.
    Pass the flow identifier, execution context, and the number of records returned.
    Returns allow=False when the record count would exceed per-run or hourly ceilings.
    """

    @staticmethod
    def check(flow_sys_id: str, flow_name: str, execution_id: str,
              record_count: int,
              credits_per_record: int = NOW_ASSIST_CREDITS_PER_FOREACH_ITEM) -> dict:
        now = time.time()
        hour_ago = now - 3600
        proposed_credits = record_count * credits_per_record

        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                if record_count > MAX_RECORDS_PER_RUN:
                    return {
                        "allow": False,
                        "reason": "per_run_record_ceiling",
                        "flow_name": flow_name,
                        "record_count": record_count,
                        "ceiling": MAX_RECORDS_PER_RUN,
                        "proposed_credits": proposed_credits,
                        "message": (
                            f"Flow {flow_name!r} ForEach would process {record_count} records "
                            f"× {credits_per_record} Now Assist credit(s) = {proposed_credits} credits. "
                            f"Per-run ceiling is {MAX_RECORDS_PER_RUN} records. "
                            "Add a more selective condition to the Look Up Records action, "
                            "or split processing into paginated batches across multiple "
                            "scheduled flow executions with an offset field."
                        ),
                    }

                hourly_runs = conn.execute(
                    "SELECT COUNT(*) FROM flow_run_log "
                    "WHERE flow_sys_id = ? AND started_at > ?",
                    (flow_sys_id, hour_ago)
                ).fetchone()[0]

                if hourly_runs >= MAX_FLOW_EXECUTIONS_PER_HOUR:
                    return {
                        "allow": False,
                        "reason": "hourly_execution_ceiling",
                        "flow_name": flow_name,
                        "executions_in_last_hour": hourly_runs,
                        "ceiling": MAX_FLOW_EXECUTIONS_PER_HOUR,
                        "proposed_credits": proposed_credits,
                        "message": (
                            f"Flow {flow_name!r} has executed {hourly_runs} times in the last hour "
                            f"(ceiling: {MAX_FLOW_EXECUTIONS_PER_HOUR}). "
                            f"Each execution approves up to {MAX_RECORDS_PER_RUN} Now Assist credits. "
                            "Verify that the flow trigger condition is not firing on every record "
                            "update across the entire table rather than a targeted subset."
                        ),
                    }

                conn.execute(
                    "INSERT INTO flow_run_log "
                    "(flow_sys_id, flow_name, execution_id, record_count, credits_approved, started_at) "
                    "VALUES (?, ?, ?, ?, ?, ?)",
                    (flow_sys_id, flow_name, execution_id, record_count, proposed_credits, now)
                )
                return {
                    "allow": True,
                    "flow_name": flow_name,
                    "record_count": record_count,
                    "approved_credits": proposed_credits,
                    "hourly_executions": hourly_runs + 1,
                }

@app.route("/guard/flow-foreach", methods=["POST"])
def flow_foreach_guard():
    data = request.get_json(force=True)
    result = FlowForEachGuard.check(
        flow_sys_id=data.get("flow_sys_id", ""),
        flow_name=data.get("flow_name", ""),
        execution_id=data.get("execution_id", ""),
        record_count=int(data.get("record_count", 0)),
        credits_per_record=int(data.get("credits_per_record", NOW_ASSIST_CREDITS_PER_FOREACH_ITEM)),
    )
    return jsonify(result), 200 if result["allow"] else 429

if __name__ == "__main__":
    init_db()
    app.run(port=8097)

Wire this as an "Outbound HTTP" step in Flow Designer immediately after the "Look Up Records" action and before the ForEach action. Pass the flow's sys_id (available via the Flow Designer context variable fd_data.flow_sys_id or a static value set at design time), a human-readable flow name, the current execution context ID, and the count of records returned by "Look Up Records" (use the length pill from the Look Up Records output). Add a "Flow Decision" (if/else) after the HTTP step: if the response body's allow field is false, branch to a "Create Record" action that logs the blocked batch in a custom u_ai_batch_log table and exits the flow. If allow is true, continue into the ForEach. This ensures that bulk trigger events producing large record sets are blocked before a single Now Assist credit is consumed by the loop body.

Failure Mode 3 — Cross-Table Trigger Cascade

ServiceNow's platform is built around interconnected tables with referential relationships: Incidents link to Problems via the problem_id field; Problems link to Changes via change_request; Tasks link to parent records via parent. Flow Designer flows trigger on events scoped to a specific table, but their actions can update records on any related table. When two flows each watch a different table and each update a field on the other's watched table, they form a mutual trigger cycle.

The canonical ServiceNow cascade is the Incident-Problem-Change triangle. A team builds three separate Now Assist-enabled flows: one for Incident (triggered on Incident update, generates an AI impact assessment and writes it to the linked Problem's u_ai_incident_impact field); one for Problem (triggered on Problem update, generates an AI root-cause analysis and writes it to linked Incidents' u_ai_root_cause fields); one for Change (triggered on Change update, generates an AI risk assessment and writes it to linked Problems' u_ai_change_risk field). Each flow is independently reasonable. Together, updating an Incident fires the Incident flow, which updates the Problem, which fires the Problem flow, which updates back to the Incident, which re-fires the Incident flow.

The cascade is uniquely dangerous because no single flow's execution history reveals the loop. Looking at the Incident flow's execution log, each run appears to have been triggered by a legitimate Incident update — because it was, just by the Problem flow on the previous cycle. The loop is only visible when correlating execution timestamps across all three flows against a single root trigger event. Without cross-flow correlation, teams diagnose the credit burn as "unexpected Incident update volume" and never connect it to the cross-table write pattern.

import time
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "sn_cascade_guard.db"

MAX_CASCADE_DEPTH = 3
CASCADE_WINDOW_SECONDS = 300

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS cascade_writes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                root_trigger_id TEXT NOT NULL,
                source_table TEXT NOT NULL,
                source_sys_id TEXT NOT NULL,
                target_table TEXT NOT NULL,
                target_sys_id TEXT NOT NULL,
                target_field TEXT NOT NULL,
                cascade_depth INTEGER DEFAULT 1,
                written_at REAL
            )
        """)
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_root_trigger "
            "ON cascade_writes (root_trigger_id, written_at)"
        )
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_target_write "
            "ON cascade_writes (root_trigger_id, target_table, target_sys_id, target_field)"
        )

class CrossTableCascadeGuard:
    """
    Call check() before writing Now Assist output to a record on a different table.
    Pass the root_trigger_id from the originating trigger event (store it in a Flow
    Designer variable at flow start and pass it through all subflow calls).
    Returns allow=False when the cascade has already written to this target record
    field in the current window, or when cascade depth exceeds the ceiling.
    """

    @staticmethod
    def check(root_trigger_id: str, source_table: str, source_sys_id: str,
              target_table: str, target_sys_id: str, target_field: str) -> dict:
        now = time.time()
        window_start = now - CASCADE_WINDOW_SECONDS

        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute(
                    "DELETE FROM cascade_writes WHERE written_at < ?",
                    (window_start,)
                )

                existing_writes = conn.execute(
                    "SELECT COUNT(*) FROM cascade_writes "
                    "WHERE root_trigger_id = ? AND written_at > ?",
                    (root_trigger_id, window_start)
                ).fetchone()[0]

                duplicate = conn.execute(
                    "SELECT cascade_depth, written_at FROM cascade_writes "
                    "WHERE root_trigger_id = ? AND target_table = ? "
                    "AND target_sys_id = ? AND target_field = ? AND written_at > ?",
                    (root_trigger_id, target_table, target_sys_id, target_field, window_start)
                ).fetchone()

                if duplicate:
                    return {
                        "allow": False,
                        "reason": "cross_table_duplicate_write",
                        "root_trigger_id": root_trigger_id,
                        "target_table": target_table,
                        "target_sys_id": target_sys_id,
                        "target_field": target_field,
                        "first_written_seconds_ago": int(now - duplicate[1]),
                        "cascade_writes_in_window": existing_writes,
                        "message": (
                            f"Root trigger {root_trigger_id!r} has already written to "
                            f"{target_table}.{target_field} on record {target_sys_id} "
                            f"({int(now - duplicate[1])}s ago). "
                            "This write is part of a cross-table cascade loop. "
                            "Blocking to prevent circular Now Assist invocations. "
                            f"Total writes from this trigger in the last "
                            f"{CASCADE_WINDOW_SECONDS}s: {existing_writes}."
                        ),
                    }

                if existing_writes >= MAX_CASCADE_DEPTH:
                    return {
                        "allow": False,
                        "reason": "cascade_depth_ceiling",
                        "root_trigger_id": root_trigger_id,
                        "cascade_writes_in_window": existing_writes,
                        "ceiling": MAX_CASCADE_DEPTH,
                        "message": (
                            f"Root trigger {root_trigger_id!r} has produced "
                            f"{existing_writes} cross-table Now Assist writes "
                            f"in the last {CASCADE_WINDOW_SECONDS}s "
                            f"(ceiling: {MAX_CASCADE_DEPTH}). "
                            "Blocking further cascade propagation. "
                            "Review the flow chain for circular table update dependencies."
                        ),
                    }

                cascade_depth = existing_writes + 1
                conn.execute(
                    "INSERT INTO cascade_writes "
                    "(root_trigger_id, source_table, source_sys_id, target_table, "
                    "target_sys_id, target_field, cascade_depth, written_at) "
                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (root_trigger_id, source_table, source_sys_id, target_table,
                     target_sys_id, target_field, cascade_depth, now)
                )
                return {
                    "allow": True,
                    "root_trigger_id": root_trigger_id,
                    "cascade_depth": cascade_depth,
                    "target_table": target_table,
                    "target_sys_id": target_sys_id,
                    "target_field": target_field,
                }

@app.route("/guard/cascade", methods=["POST"])
def cascade_guard():
    data = request.get_json(force=True)
    result = CrossTableCascadeGuard.check(
        root_trigger_id=data.get("root_trigger_id", ""),
        source_table=data.get("source_table", ""),
        source_sys_id=data.get("source_sys_id", ""),
        target_table=data.get("target_table", ""),
        target_sys_id=data.get("target_sys_id", ""),
        target_field=data.get("target_field", ""),
    )
    return jsonify(result), 200 if result["allow"] else 429

if __name__ == "__main__":
    init_db()
    app.run(port=8098)

Assign a root trigger ID at the very start of every Now Assist-enabled flow: create a flow variable root_trigger_id and set it to GlideGuid.generate() as the first action, before any Now Assist calls or cross-table writes. When this flow calls subflows or executes actions that update related table records, pass root_trigger_id as a subflow input parameter. In each subflow, call the /guard/cascade endpoint before writing Now Assist output to a linked record — pass the inherited root_trigger_id, the source table and sys_id, and the target table, sys_id, and field name. A 429 response means this write is part of a cascade already tracked from the same root trigger: skip the Now Assist call and the field update entirely. The cascade_writes_in_window counter in every response is the cross-flow observability signal that reveals cascade depth in real time.

Failure Mode 4 — Scheduled Job Concurrency Overlap

ServiceNow's Scheduled Script Execution mechanism runs server-side scripts on a defined schedule — hourly, nightly, weekly. Scheduled jobs are a common pattern for batch AI enrichment: process all open Incidents with no AI summary, call Now Assist for each, write the summary, mark the record as processed. The job runs nightly at midnight and finishes by 2 AM on a typical night.

The concurrency failure mode emerges as the incident volume grows. A ServiceNow instance that handled 500 incidents per night during initial deployment now handles 8,000 incidents per night after two years of growth. The nightly batch job that completed in 90 minutes now takes 26 hours. The job is still scheduled to run nightly. At midnight on day 2, the previous night's job has not finished — it is still processing records from 26 hours ago. ServiceNow starts a new instance of the scheduled job. Both instances query for records with no AI summary (u_ai_summary is empty). At the time of the second query, most records still have no summary because the first instance has only processed a fraction. Both instances begin calling Now Assist on the same unprocessed records. Credit consumption doubles. Both instances write summaries back to the same records, triggering business rules on each write, which in turn may trigger further Now Assist calls.

The concurrency problem is exacerbated by ServiceNow's lack of a native row-level locking mechanism for Scheduled Script Execution — unlike database-level row locking in a transactional system, two simultaneously running GlideRecord queries on the same filter condition each return an independent cursor over the same matching rows, with no mutual exclusion.

import time
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "sn_job_lock.db"

LOCK_MAX_SECONDS = 28800  # 8 hours — 2× expected max batch job duration

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS job_locks (
                job_name TEXT PRIMARY KEY,
                run_id TEXT NOT NULL,
                acquired_at REAL NOT NULL,
                records_estimated INTEGER DEFAULT 0,
                records_processed INTEGER DEFAULT 0,
                released_at REAL
            )
        """)

class ScheduledJobLock:

    @staticmethod
    def acquire(job_name: str, run_id: str, records_estimated: int = 0) -> dict:
        now = time.time()
        expired_before = now - LOCK_MAX_SECONDS

        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                existing = conn.execute(
                    "SELECT run_id, acquired_at, released_at, records_processed "
                    "FROM job_locks WHERE job_name = ?",
                    (job_name,)
                ).fetchone()

                if existing:
                    held_run_id, acquired_at, released_at, records_processed = existing
                    is_released = released_at is not None
                    is_expired = acquired_at < expired_before

                    if not is_released and not is_expired:
                        running_for = int(now - acquired_at)
                        return {
                            "acquired": False,
                            "reason": "job_already_running",
                            "job_name": job_name,
                            "held_by_run_id": held_run_id,
                            "running_for_seconds": running_for,
                            "records_processed_so_far": records_processed,
                            "lock_watchdog_expires_in": int(LOCK_MAX_SECONDS - running_for),
                            "message": (
                                f"Scheduled job {job_name!r} is already running "
                                f"(run {held_run_id!r}, started {running_for}s ago, "
                                f"{records_processed} records processed so far). "
                                "This instance will exit without calling Now Assist "
                                "to prevent duplicate credit consumption on the same records. "
                                f"Lock watchdog expires in {int(LOCK_MAX_SECONDS - running_for)}s "
                                "if the previous instance crashes."
                            ),
                        }

                conn.execute(
                    "INSERT OR REPLACE INTO job_locks "
                    "(job_name, run_id, acquired_at, records_estimated, "
                    "records_processed, released_at) "
                    "VALUES (?, ?, ?, ?, 0, NULL)",
                    (job_name, run_id, now, records_estimated)
                )
                return {
                    "acquired": True,
                    "job_name": job_name,
                    "run_id": run_id,
                    "lock_expires_at": now + LOCK_MAX_SECONDS,
                    "records_estimated": records_estimated,
                }

    @staticmethod
    def progress(job_name: str, run_id: str, records_processed: int) -> dict:
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute(
                    "UPDATE job_locks SET records_processed = ? "
                    "WHERE job_name = ? AND run_id = ?",
                    (records_processed, job_name, run_id)
                )
                return {"updated": True, "job_name": job_name, "records_processed": records_processed}

    @staticmethod
    def release(job_name: str, run_id: str, records_processed: int = 0) -> dict:
        now = time.time()
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute(
                    "UPDATE job_locks "
                    "SET released_at = ?, records_processed = ? "
                    "WHERE job_name = ? AND run_id = ?",
                    (now, records_processed, job_name, run_id)
                )
                return {
                    "released": True,
                    "job_name": job_name,
                    "run_id": run_id,
                    "records_processed": records_processed,
                }

@app.route("/lock/job/acquire", methods=["POST"])
def acquire_job_lock():
    data = request.get_json(force=True)
    result = ScheduledJobLock.acquire(
        job_name=data.get("job_name", ""),
        run_id=data.get("run_id", ""),
        records_estimated=int(data.get("records_estimated", 0)),
    )
    return jsonify(result), 200 if result["acquired"] else 409

@app.route("/lock/job/progress", methods=["POST"])
def update_job_progress():
    data = request.get_json(force=True)
    result = ScheduledJobLock.progress(
        job_name=data.get("job_name", ""),
        run_id=data.get("run_id", ""),
        records_processed=int(data.get("records_processed", 0)),
    )
    return jsonify(result), 200

@app.route("/lock/job/release", methods=["POST"])
def release_job_lock():
    data = request.get_json(force=True)
    result = ScheduledJobLock.release(
        job_name=data.get("job_name", ""),
        run_id=data.get("run_id", ""),
        records_processed=int(data.get("records_processed", 0)),
    )
    return jsonify(result), 200

if __name__ == "__main__":
    init_db()
    app.run(port=8099)

At the start of the Scheduled Script Execution script, call /lock/job/acquire via sn_ws.RESTMessageV2, passing the job name and a generated run ID (gs.generateGUID()). If the response status is 409, log the message to the system log (gs.warn()) and exit the script immediately with return — no GlideRecord query, no Now Assist calls. If the response is 200 and acquired is true, proceed with the batch processing loop. Inside the loop, call /lock/job/progress every 100 records to update the processed count — this gives operational visibility into how far the job has progressed when monitoring the lock endpoint. At the end of the script — including inside any catch block — call /lock/job/release with the final record count. Wrap the release call in a try/catch that only logs failure, never re-throws, so a release failure does not mask the original job completion status. The LOCK_MAX_SECONDS watchdog ensures that a job that crashes mid-execution without reaching the release call automatically frees the lock after 8 hours, preventing subsequent nights from being permanently blocked.

State Table

Failure mode	Guard class	Ceiling / trigger	What to watch
Business rule recursion AI field write re-triggers the same business rule on the same record	BusinessRuleRecursionGuard	Hash match = skip current.update() (idempotency)	recursive_blocks_total per record; spikes on specific sys_ids reveal high-churn records with unstable AI output
Flow Designer ForEach fan-out Unbounded Look Up Records result set × Now Assist per iteration	FlowForEachGuard	50 records per run, 10 executions per hour	Blocked runs in u_ai_batch_log; persistent blocks = query filter too broad
Cross-table trigger cascade Flow A writes to Table B, triggering Flow B which writes back to Table A	CrossTableCascadeGuard	3 cross-table writes per root trigger per 5-minute window	cascade_writes_in_window per root trigger; values >3 indicate active circular dependency
Scheduled job concurrency New job instance fires before previous completes; both process same records	ScheduledJobLock	1 concurrent instance; watchdog at 8 hours	409 response rate on acquire endpoint; check records_processed_so_far to estimate when previous instance will finish

Checklist Before Going Live

1. Audit every "after" business rule on tables where Now Assist writes AI-generated fields. For each rule with a current.update() or current.setValue() + current.update() pattern that also calls a Now Assist Skill, add the hash idempotency check before the write. Prioritize tables with high update frequency in your environment — Incident, Task, and Change are the most common sources of recursive business rule invocations.
2. Add a record count pre-flight check to every Flow Designer flow that contains a ForEach loop with Now Assist actions. Use the ServiceNow "Aggregate" action (type: COUNT, no GroupBy) on the same filter as your Look Up Records action to get the count before the full query runs. If the count exceeds your per-run ceiling, branch to a notification record and exit — do not execute the Look Up Records action with the full result set.
3. Establish a root trigger ID convention for all Now Assist-enabled flows. Every flow that may write to records on related tables must generate a root trigger ID as its first action and pass it through all subflow parameters. This is the foundational requirement for the cross-table cascade guard to function — without a shared root trigger ID threaded through the flow chain, the guard cannot correlate writes from the same originating event.
4. Set Now Assist Skill temperature or determinism parameters to their most stable values for field-population use cases. The business rule recursion guard relies on hash stability — if the Now Assist Skill produces different outputs on two successive calls for the same input (due to non-zero temperature), the hash check will always allow writes and the recursion guard cannot block the loop. For structured field population (incident summaries, root cause drafts, risk assessments), configure the Now Assist Skill with the lowest available temperature or enable deterministic mode where the platform offers it.
5. Monitor the Now Assist usage dashboard in ServiceNow's platform analytics against your expected baseline. Set a daily credit consumption alert at 60% of your monthly allocation — not 80%, because a single runaway batch job on a high-volume table can consume the remaining 40% in one hour before the next monitoring check. The Now Assist usage report in the Platform Analytics workspace shows credit consumption by Skill, by flow, and by table.
6. Design all batch AI enrichment scheduled jobs with a pagination cursor rather than a single bulk query. Instead of querying all records with u_ai_summary is empty in one execution, store a last_processed_sys_id cursor in a system property. Each job run processes the next N records starting from the cursor and updates the cursor at completion. This bounds per-execution credit consumption to N records regardless of total queue depth, and the concurrency lock guard's records_processed counter gives you real-time progress visibility.
7. Test cross-table write patterns in a sub-production instance with Now Assist credit metering enabled before deploying to production. ServiceNow's development instances do not always replicate production credit metering behavior, but you can observe the cascade pattern by checking the System Log after triggering a representative update event. If the log shows Now Assist invocations for both the Incident flow and the Problem flow within the same 5-second window from a single Incident update, the cascade is already present and the guard must be added before production deployment.

FAQ

How does ServiceNow Now Assist billing differ from other Microsoft or Salesforce AI platforms we've already deployed?

Now Assist uses a credit-per-generative-AI-invocation model where each Now Assist Skill call, each "Generate Now Assist Text" Flow Designer action, and each AI-assisted field population consumes credits from your organization's Now Assist entitlement. This is structurally similar to Power Automate AI Builder's per-credit model but distinct from Copilot Studio's per-message model and Salesforce Agentforce's per-conversation model. The ServiceNow-specific risk is that the platform's event-driven execution model (business rules firing on GlideRecord writes) creates implicit invocation chains that are not visible in any single flow's execution history — the billing event is hidden inside a platform mechanism that most administrators monitor only for performance, not for AI credit consumption.

Does ServiceNow's native business rule protection against infinite loops make the recursion guard redundant?

No. ServiceNow's native business rule loop protection terminates a chain after 30 recursive executions on the same record within a single transaction. This is a last-resort platform safety net, not a cost control mechanism. At one Now Assist credit per recursion level, the platform allows 30 credits to be consumed before stopping the loop — and this is per-record, so 100 simultaneously updated records generate up to 3,000 unintended credits before the native protection trips. The hash-based idempotency guard stops the recursion at the second invocation (blocking any Now Assist call when output is unchanged), keeping consumption at 1 credit per legitimate trigger rather than 30. The native platform protection and the external guard are complementary, not redundant.

Can we pass the root trigger ID through ServiceNow's native Flow Designer parameters without custom code?

Yes. Flow Designer supports input parameters on flows and subflows — you define them in the flow's properties panel as "Input Variables." When a parent flow calls a subflow via the "Call Subflow" action, you pass the parent's root_trigger_id variable to the subflow's corresponding input variable. The subflow uses this variable in its guard endpoint call and any further downstream subflow invocations. This requires no custom code — only parameter wiring in the Flow Designer visual editor. The root trigger ID variable must be defined on every flow and subflow in the chain; flows that do not receive it as an input cannot participate in the cascade guard and should be treated as potential cascade entry points requiring separate audit.

We have hundreds of ServiceNow flows. How do we identify which ones are at risk without reviewing each one individually?

Query the ServiceNow Flow Designer table sys_hub_flow and the Flow Designer action table sys_hub_action_type_base for flows that contain both a "Look Up Records" action and a "Generate Now Assist Text" action (or any Now Assist Skill invocation action). Flows matching both criteria in the same flow body are candidates for the ForEach fan-out risk. For business rule recursion, query sys_script for active business rules on high-volume tables (Incident, Task, Change, Problem) that contain the string "NowAssist" or the Now Assist Skill API identifier. For cross-table cascade risk, identify flows whose actions write to a related table field that is itself a trigger condition for another flow — this requires mapping flow trigger conditions against flow action targets, which can be scripted using the ServiceNow REST API against the Flow Designer tables.

How do I integrate RunGuard's SDK with ServiceNow Now Assist flows?

Deploy RunGuard on a server accessible from ServiceNow's HTTP connector allowlist. Use RunGuard's BudgetTracker as the backing engine for the ForEach fan-out guard — call tracker.record(session_id=execution_id, tokens=record_count) to log approved credit reservations and tracker.check(session_id=execution_id) to enforce per-run ceilings. Use RunGuard's LoopDetector for the business rule recursion detection — each Now Assist call from the business rule is logged with detector.record(session_id=sys_id, tool="now_assist_skill_name"), and repeated identical tool calls on the same sys_id within a short window trip the detector before the next invocation is allowed. Call RunGuard's endpoints from ServiceNow's Outbound HTTP (sn_ws.RESTMessageV2) actions in Flow Designer or from the business rule's server-side JavaScript. Install RunGuard with pip install runguard and host it on any internet-accessible endpoint that ServiceNow can reach via the MID Server or direct HTTP outbound connection.