diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml
index 67c06d0..1872e5c 100644
--- a/.github/workflows/pr.yml
+++ b/.github/workflows/pr.yml
@@ -98,12 +98,10 @@ jobs:
         uses: Ortus-Solutions/setup-commandbox@v2.0.1
 
       - name: Install CFFormat
-        run: box install commandbox-cfformat
+        # 0.22.1 adds whitespace before unquoted struct-key colons on every pass.
+        run: box install commandbox-cfformat@0.21.0
 
-      - name: Run CFFormat
-        run: box run-script format
-
-      - name: Commit Format Changes
-        uses: stefanzweifel/git-auto-commit-action@v5.2.0
-        with:
-          commit_message: Apply cfformat changes
+      - name: Check CFFormat
+        # Report changes for the author to apply locally; never create another
+        # workflow-triggering commit from the formatter job.
+        run: box run-script format:check
diff --git a/ModuleConfig.cfc b/ModuleConfig.cfc
index f2104bd..557af47 100644
--- a/ModuleConfig.cfc
+++ b/ModuleConfig.cfc
@@ -5,7 +5,7 @@ component {
 	this.webUrl = "https://github.com/coldbox-modules/cbq";
 	this.cfmapping = "cbq";
 
-	variables.customInterceptionPointMethodNames = "onCBQJobAdded,onCBQJobMarshalled,onCBQJobComplete,onCBQJobException,onCBQJobFailed";
+	variables.customInterceptionPointMethodNames = "onCBQJobAdded,onCBQJobMarshalled,onCBQJobComplete,onCBQJobException,onCBQJobFailed,onCBQJobAttemptScheduled,onCBQJobAttemptFinished,onCBQJobExecutionStarted,onCBQJobExecutionExited,onCBQJobPublished,onCBQJobPublishException";
 
 	function configure() {
 		settings = {
diff --git a/README.md b/README.md
index 6150ece..c7e7e98 100644
--- a/README.md
+++ b/README.md
@@ -301,3 +301,11 @@ With a larger batch size, each job is still announced separately and becomes its
 The DB provider writes at most 100 rows (500 bindings) per insert. Other providers continue through their ordinary `push` implementation unless they implement `pushMany(entries)`. Entries contain the ordinary `push` arguments: `queueName`, `job`, and optionally `delay` and `attempts`. `DBProvider.pushMany` also bounds direct calls to 100 rows per insert. It uses the configured table, query options and datasource.
 
 The caller owns transaction boundaries. Wrap the dispatch with related database changes when all chunks must commit or roll back together. Without a surrounding transaction, a later event, serialization or insert failure can leave earlier chunks persisted.
+
+## Observational execution hooks
+
+Publishing announces onCBQJobPublished or onCBQJobPublishException after the provider boundary, including bulk dispatch. Added observational hooks isolate observer failures and preserve the original enqueue error. onCBQJobAdded retains its existing contract.
+
+Every attempt announces onCBQJobAttemptScheduled, onCBQJobExecutionStarted, onCBQJobExecutionExited, and onCBQJobAttemptFinished. Data contains job, immutable executionId, and attempt. Finished additionally contains status: ok, released, internal_error, cancelled, or deadline_exceeded. Scheduled precedes dispatch, Started/Exited bracket the actual worker body in try/finally, while Finished can execute on another thread. Consumers must key handles by executionId, restore context in Exited, and finish atomically. A timed-out worker can exit later. Existing completion/error hooks retain their behavior.
+
+Focused coverage: synchronous observer failures, bulk enqueue failures, and a real asynchronous timeout. Comprehensive retry/chain/batch/provider and engine matrices remain separate checks.
diff --git a/box.json b/box.json
index 93458a8..8610161 100644
--- a/box.json
+++ b/box.json
@@ -9,34 +9,34 @@
         "type":"git",
         "URL":"https://github.com/coldbox-modules/cbq"
     },
-    "bugs":"https://github.com/coldbox-modules/cbq/issues",
-    "slug":"cbq",
-    "shortDescription":"A protocol-based queueing system for ColdBox",
-    "description":"A protocol-based queueing system for ColdBox",
-    "type":"modules",
-    "dependencies":{
-        "qb":"^13.0.0",
-        "str":"^4.0.0"
+    "bugs": "https://github.com/coldbox-modules/cbq/issues",
+    "slug": "cbq",
+    "shortDescription": "A protocol-based queueing system for ColdBox",
+    "description": "A protocol-based queueing system for ColdBox",
+    "type": "modules",
+    "dependencies": {
+        "qb": "^13.0.0",
+        "str": "^4.0.0"
     },
-    "devDependencies":{
-        "testbox":"^6.0.0",
-        "coldbox":"^7.0.0",
-        "cfmigrations":"^5.0.0"
+    "devDependencies": {
+        "testbox": "^6.0.0",
+        "coldbox": "^7.0.0",
+        "cfmigrations": "^5.0.0"
     },
-    "installPaths":{
-        "testbox":"testbox/",
-        "coldbox":"tests/resources/app/coldbox/",
-        "qb":"modules/qb/",
-        "str":"modules/str/",
-        "cfmigrations":"modules/cfmigrations/"
+    "installPaths": {
+        "testbox": "testbox/",
+        "coldbox": "tests/resources/app/coldbox/",
+        "qb": "modules/qb/",
+        "str": "modules/str/",
+        "cfmigrations": "modules/cfmigrations/"
     },
-    "scripts":{
-        "format":"cfformat run ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc --overwrite",
-        "format:check":"cfformat check ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc --verbose",
-        "format:watch":"cfformat watch ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc",
-        "install:2021":"cfpm install document,feed,zip"
+    "scripts": {
+        "format": "cfformat run ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc --overwrite",
+        "format:check": "cfformat check ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc --verbose",
+        "format:watch": "cfformat watch ModuleConfig.cfc,models/**/*.cfc,interceptors/**/*.cfc,tests/specs/**/*.cfc,tests/resources/app/handlers/**/*.cfc,tests/resources/app/config/**/*.cfc,tests/resources/app/models/**/*.cfc",
+        "install:2021": "cfpm install document,feed,zip"
     },
-    "ignore":[
+    "ignore": [
         "**/.*",
         "test",
         "tests"
diff --git a/models/Jobs/Dispatcher.cfc b/models/Jobs/Dispatcher.cfc
index 62a10b3..5cb5283 100644
--- a/models/Jobs/Dispatcher.cfc
+++ b/models/Jobs/Dispatcher.cfc
@@ -2,6 +2,7 @@ component singleton accessors="true" {
 
 	property name="interceptorService" inject="box:interceptorService";
 	property name="config" inject="Config@cbq";
+	property name="log" inject="logbox:logger:{this}";
 
 	public Dispatcher function dispatch( required any job ) {
 		var connectionName = arguments.job.getConnection();
@@ -22,11 +23,26 @@ component singleton accessors="true" {
 		);
 
 		arguments.job.setCurrentAttempt( 0 );
-		connection.push(
-			queueName = queueName,
-			job = arguments.job,
-			delay = delay,
-			attempts = 0
+		try {
+			connection.push(
+				queueName = queueName,
+				job = arguments.job,
+				delay = delay,
+				attempts = 0
+			);
+		} catch ( any failure ) {
+			announcePublishResult(
+				"onCBQJobPublishException",
+				arguments.job,
+				connection,
+				failure
+			);
+			rethrow;
+		}
+		announcePublishResult(
+			"onCBQJobPublished",
+			arguments.job,
+			connection
 		);
 
 		return this;
@@ -61,11 +77,11 @@ component singleton accessors="true" {
 				"attempts" : 0
 			};
 			if ( arguments.batchSize == 1 ) {
-				connection.push( argumentCollection = entry );
+				publishMany( connection, [ entry ] );
 			} else {
 				pending.append( entry );
 				if ( pending.len() == arguments.batchSize ) {
-					connection.pushMany( pending );
+					publishMany( connection, pending );
 					pending = [];
 				}
 			}
@@ -78,4 +94,55 @@ component singleton accessors="true" {
 		return this;
 	}
 
+	private function publishMany( required any connection, required array entries ) {
+		try {
+			if ( arguments.entries.len() == 1 ) {
+				arguments.connection.push( argumentCollection = arguments.entries[ 1 ] );
+			} else {
+				arguments.connection.pushMany( arguments.entries );
+			}
+		} catch ( any failure ) {
+			for ( var entry in arguments.entries ) {
+				announcePublishResult(
+					"onCBQJobPublishException",
+					entry.job,
+					arguments.connection,
+					failure
+				);
+			}
+			rethrow;
+		}
+		for ( var entry in arguments.entries ) {
+			announcePublishResult(
+				"onCBQJobPublished",
+				entry.job,
+				arguments.connection
+			);
+		}
+	}
+
+	private function announcePublishResult(
+		required string state,
+		required any job,
+		required any connection,
+		any exception
+	) {
+		try {
+			variables.interceptorService.announce(
+				arguments.state,
+				{
+					"job" : arguments.job,
+					"connection" : arguments.connection,
+					"exception" : arguments.exception ?: javacast( "null", "" )
+				}
+			);
+		} catch ( any observerFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				arguments.state,
+				observerFailure,
+				variables.log ?: javacast( "null", "" )
+			);
+		}
+	}
+
 }
diff --git a/models/Providers/AbstractQueueProvider.cfc b/models/Providers/AbstractQueueProvider.cfc
index 8870c63..61b3eae 100644
--- a/models/Providers/AbstractQueueProvider.cfc
+++ b/models/Providers/AbstractQueueProvider.cfc
@@ -105,27 +105,43 @@ component accessors="true" {
 		function afterJobHook
 	) {
 		arguments.job.setCurrentAttempt( ( arguments.job.getCurrentAttempt() ?: 0 ) + 1 );
+		var observation = {
+			"job" : arguments.job,
+			"executionId" : createUUID(),
+			"attempt" : arguments.job.getCurrentAttempt()
+		};
+		announceExecutionObservation( "onCBQJobAttemptScheduled", observation );
 		return variables.async
 			.newFuture( function() {
-				if ( variables.log.canDebug() ) {
-					variables.log.debug( "Marshaling job ###job.getId()#", job.getMemento() );
-				}
+				announceExecutionObservation( "onCBQJobExecutionStarted", observation );
+				try {
+					if ( variables.log.canDebug() ) {
+						variables.log.debug( "Marshaling job ###job.getId()#", job.getMemento() );
+					}
 
-				beforeJobRun( job );
-				if ( structKeyExists( job, "before" ) ) {
-					job.before();
-				}
+					beforeJobRun( job );
+					if ( structKeyExists( job, "before" ) ) {
+						job.before();
+					}
 
-				variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } );
+					variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } );
 
-				if ( variables.log.canDebug() ) {
-					variables.log.debug( "Running job ###job.getId()#", job.getMemento() );
-				}
+					if ( variables.log.canDebug() ) {
+						variables.log.debug( "Running job ###job.getId()#", job.getMemento() );
+					}
 
-				return job.handle();
+					// These paired hooks bracket the actual worker execution, including failures.
+					// Completion callbacks may run on a different thread.
+					return job.handle();
+				} finally {
+					announceExecutionObservation( "onCBQJobExecutionExited", observation );
+				}
 			}, arguments.pool.getExecutor() )
 			.orTimeout( getTimeoutForJob( arguments.job, arguments.pool ), "seconds" )
 			.then( function( result ) {
+				var completed = structCopy( observation );
+				completed.status = job.getIsReleased() ? "released" : "ok";
+				announceExecutionObservation( "onCBQJobAttemptFinished", completed );
 				if ( job.getIsReleased() ) {
 					variables.log.debug( "Job [#job.getId()#] requested manual release." );
 
@@ -180,6 +196,11 @@ component accessors="true" {
 				}
 			} )
 			.onException( function( e ) {
+				var failed = structCopy( observation );
+				failed.status = job.getIsCancelled() ? "cancelled" : (
+					!isNull( e ) && findNoCase( "Timeout", e.toString() ) ? "deadline_exceeded" : "internal_error"
+				);
+				announceExecutionObservation( "onCBQJobAttemptFinished", failed );
 				try {
 					// log failed job
 					if ( !isNull( e ) && "java.util.concurrent.CompletionException" == e.getClass().getName() ) {
@@ -264,7 +285,12 @@ component accessors="true" {
 										"releaseException" : releaseException
 									}
 								);
-							} catch ( any ignored ) {
+							} catch ( any diagnosticFailure ) {
+								new cbq.models.Support.FailureDiagnostics().report(
+									"releaseJob diagnostic",
+									diagnosticFailure,
+									variables.log ?: javacast( "null", "" )
+								);
 							}
 							markJobFailed(
 								job,
@@ -293,11 +319,21 @@ component accessors="true" {
 								"handlerException" : outerException
 							}
 						);
-					} catch ( any ignored ) {
+					} catch ( any diagnosticFailure ) {
+						new cbq.models.Support.FailureDiagnostics().report(
+							"onException handler diagnostic",
+							diagnosticFailure,
+							variables.log ?: javacast( "null", "" )
+						);
 					}
 					try {
 						forceFailJob( job.getId(), pool, job );
-					} catch ( any ignored ) {
+					} catch ( any diagnosticFailure ) {
+						new cbq.models.Support.FailureDiagnostics().report(
+							"forceFailJob fallback",
+							diagnosticFailure,
+							variables.log ?: javacast( "null", "" )
+						);
 					}
 				}
 
@@ -409,7 +445,12 @@ component accessors="true" {
 					arguments.pool,
 					arguments.job
 				);
-			} catch ( any ignored ) {
+			} catch ( any diagnosticFailure ) {
+				new cbq.models.Support.FailureDiagnostics().report(
+					"afterJobFailed forceFailJob fallback",
+					diagnosticFailure,
+					variables.log ?: javacast( "null", "" )
+				);
 			}
 		}
 
@@ -439,7 +480,12 @@ component accessors="true" {
 					{ "exception" : arguments.exception }
 				);
 			}
-		} catch ( any ignored ) {
+		} catch ( any diagnosticFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				"side effect diagnostic",
+				diagnosticFailure,
+				variables.log ?: javacast( "null", "" )
+			);
 		}
 	}
 
@@ -595,4 +641,16 @@ component accessors="true" {
 		return;
 	}
 
+	function announceExecutionObservation( required string state, required struct data ) {
+		try {
+			variables.interceptorService.announce( arguments.state, arguments.data );
+		} catch ( any observerFailure ) {
+			new cbq.models.Support.FailureDiagnostics().report(
+				arguments.state,
+				observerFailure,
+				variables.log ?: javacast( "null", "" )
+			);
+		}
+	}
+
 }
diff --git a/models/Providers/SyncProvider.cfc b/models/Providers/SyncProvider.cfc
index c7adaf9..17edf60 100644
--- a/models/Providers/SyncProvider.cfc
+++ b/models/Providers/SyncProvider.cfc
@@ -59,6 +59,13 @@ component accessors="true" extends="AbstractQueueProvider" {
 	}
 
 	public void function marshalJob( required AbstractJob job, required WorkerPool pool ) {
+		var observation = {
+			"job" : arguments.job,
+			"executionId" : createUUID(),
+			"attempt" : arguments.job.getCurrentAttempt()
+		};
+		announceExecutionObservation( "onCBQJobAttemptScheduled", observation );
+		announceExecutionObservation( "onCBQJobExecutionStarted", observation );
 		try {
 			if ( variables.log.canDebug() ) {
 				// variables.log.debug( "Marshaling job ###arguments.job.getId()#", arguments.job.getMemento() );
@@ -76,6 +83,9 @@ component accessors="true" extends="AbstractQueueProvider" {
 			}
 
 			var result = arguments.job.handle();
+			var completed = structCopy( observation );
+			completed.status = job.getIsReleased() ? "released" : "ok";
+			announceExecutionObservation( "onCBQJobAttemptFinished", completed );
 
 			if ( job.getIsReleased() ) {
 				variables.log.debug( "Job [#job.getId()#] requested manual release." );
@@ -117,6 +127,9 @@ component accessors="true" extends="AbstractQueueProvider" {
 
 			ensureSuccessfulBatchJobIsRecorded( job, pool );
 		} catch ( any e ) {
+			var failed = structCopy( observation );
+			failed.status = job.getIsCancelled() ? "cancelled" : "internal_error";
+			announceExecutionObservation( "onCBQJobAttemptFinished", failed );
 			// log failed job
 			if ( log.canError() ) {
 				log.error( "Exception when running job: #e.message#" );
@@ -152,6 +165,8 @@ component accessors="true" extends="AbstractQueueProvider" {
 
 				rethrow;
 			}
+		} finally {
+			announceExecutionObservation( "onCBQJobExecutionExited", observation );
 		}
 	}
 
diff --git a/models/Support/FailureDiagnostics.cfc b/models/Support/FailureDiagnostics.cfc
new file mode 100644
index 0000000..ade4dae
--- /dev/null
+++ b/models/Support/FailureDiagnostics.cfc
@@ -0,0 +1,22 @@
+/** Diagnostics must never replace the queue outcome they are reporting. */
+component {
+
+	public void function report(
+		required string stage,
+		required any exception,
+		any logger
+	) {
+		try {
+			arguments.logger.warn( "cbq diagnostic failure at [#arguments.stage#]", arguments.exception );
+		} catch ( any loggingFailure ) {
+			// Do not include job payloads or exception messages in the fallback.
+			createObject( "java", "java.lang.System" ).err.println(
+				"cbq: [#arguments.stage#] failed; LogBox reporting also failed [#loggingFailure.type#]."
+			);
+		} finally {
+			// Even an unavailable stderr must not alter persistence, retries or cleanup.
+			return;
+		}
+	}
+
+}
diff --git a/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc b/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc
index c4b48af..47faffb 100644
--- a/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc
+++ b/tests/specs/integration/Providers/DBProviderTimeoutWatcherSpec.cfc
@@ -154,8 +154,21 @@ component extends="tests.resources.ModuleIntegrationSpec" appMapping="/app" {
 				return [];
 			}
 		);
-		prepareMock( variables.provider ).$( "newQuery", builder );
-		variables.provider.fetchPotentiallyOpenRecords( capacity = 10, pool = variables.pool );
+		// Keep the compile-only builder off the application provider used by teardown.
+		var compilationProvider = createMock( "cbq.models.Providers.DBProvider" ).setProperties( {} );
+		compilationProvider.$property(
+			"javaInstant",
+			"variables",
+			createObject( "java", "java.time.Instant" )
+		);
+		compilationProvider.$property(
+			"log",
+			"variables",
+			{ canDebug : () => false }
+		);
+		compilationProvider.$( "newQuery", builder );
+		makePublic( compilationProvider, "fetchPotentiallyOpenRecords" );
+		compilationProvider.fetchPotentiallyOpenRecords( capacity = 10, pool = variables.pool );
 
 		var sql = builder.toSQL();
 		expect( sql ).toInclude( "CASE WHEN #arguments.quotedColumn# = ? THEN 1 ELSE 2 END ASC" );
diff --git a/tests/specs/unit/ObservabilityHooksSpec.cfc b/tests/specs/unit/ObservabilityHooksSpec.cfc
new file mode 100644
index 0000000..912d696
--- /dev/null
+++ b/tests/specs/unit/ObservabilityHooksSpec.cfc
@@ -0,0 +1,394 @@
+component extends="testbox.system.BaseSpec" {
+
+	function run() {
+		describe( "observational queue boundaries", function() {
+			it( "brackets synchronous worker execution and preserves the result when observers fail", function() {
+				var states = [];
+				var diagnostics = [];
+				var result = 0;
+				var provider = createMock( "cbq.models.Providers.SyncProvider" );
+				provider.$property(
+					"log",
+					"variables",
+					{
+						canDebug : function() {
+							return false;
+						},
+						canError : function() {
+							return false;
+						},
+						warn : function( message, extraInfo ) {
+							diagnostics.append( {
+								message : message,
+								exception : extraInfo
+							} );
+						}
+					}
+				);
+				provider.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : function( state, data ) {
+							states.append( {
+								state : state,
+								id : data.executionId ?: ""
+							} );
+							if ( state == "onCBQJobComplete" ) {
+								result = data.result;
+							}
+							if ( state == "onCBQJobExecutionStarted" ) {
+								throw( "ObserverFailure" );
+							}
+						}
+					}
+				);
+				var job = createMock( "cbq.models.Jobs.AbstractJob" ).init();
+				job.setId( "synthetic" )
+					.setCurrentAttempt( 1 )
+					.setMapping( "Synthetic" );
+				job.$( "handle" )
+					.$callback( function() {
+						return 42;
+					} );
+				job.$( "isBatchJob", false );
+				var pool = createMock( "cbq.models.Workers.WorkerPool" ).init();
+				provider.marshalJob( job, pool );
+				expect( states.map( ( entry ) => entry.state ) ).toBe( [
+					"onCBQJobAttemptScheduled",
+					"onCBQJobExecutionStarted",
+					"onCBQJobMarshalled",
+					"onCBQJobAttemptFinished",
+					"onCBQJobComplete",
+					"onCBQJobExecutionExited"
+				] );
+				expect( states[ 1 ].id ).toBe( states[ 6 ].id );
+				expect( result ).toBe( 42 );
+				expect( diagnostics.len() ).toBe( 1 );
+				expect( diagnostics[ 1 ].message ).toInclude( "onCBQJobExecutionStarted" );
+			} );
+			it( "finishes a real asynchronous timeout before the worker exits and keeps a stable execution ID", function() {
+				var async = new coldbox.system.async.AsyncManager();
+				var executor = async.newExecutor(
+					"observation-test-" & createUUID(),
+					"single",
+					1
+				);
+				var states = createObject( "java", "java.util.concurrent.ConcurrentLinkedQueue" ).init();
+				var started = createObject( "java", "java.util.concurrent.CountDownLatch" ).init( 1 );
+				var exited = createObject( "java", "java.util.concurrent.CountDownLatch" ).init( 1 );
+				var provider = createMock( "cbq.models.Providers.AbstractQueueProvider" );
+				provider.$property( "async", "variables", async );
+				provider.$property(
+					"log",
+					"variables",
+					{
+						canDebug : () => false,
+						canError : () => false,
+						warn : () => {
+							throw( type = "DiagnosticFailure" );
+						},
+						debug : () => {
+						}
+					}
+				);
+				provider.$( "releaseJob", () => {
+				} );
+				provider.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : function( state, data ) {
+							if ( data.keyExists( "executionId" ) ) {
+								states.add( {
+									"state" : state,
+									"id" : data.executionId,
+									"status" : data.status ?: ""
+								} );
+							}
+							if ( state == "onCBQJobExecutionStarted" ) {
+								started.countDown();
+								throw( type = "ObserverFailure" );
+							}
+							if ( state == "onCBQJobExecutionExited" ) {
+								exited.countDown();
+								throw( type = "ObserverFailure" );
+							}
+						}
+					}
+				);
+				var job = createMock( "cbq.models.Jobs.AbstractJob" )
+					.init()
+					.setId( "synthetic" )
+					.setMapping( "Synthetic" )
+					.setTimeout( 1 )
+					.setMaxAttempts( 2 );
+				job.$( "handle" )
+					.$callback( () => {
+						sleep( 4000 );
+						return 42;
+					} );
+				job.$( "isBatchJob", false );
+				var pool = createMock( "cbq.models.Workers.WorkerPool" )
+					.init()
+					.setExecutor( executor )
+					.setMaxAttempts( 2 );
+				try {
+					provider.marshalJob( job, pool ).get( 6500 );
+					expect( started.getCount() ).toBe( 0 );
+
+					expect(
+						exited.await(
+							javacast( "long", 6 ),
+							createObject( "java", "java.util.concurrent.TimeUnit" ).SECONDS
+						)
+					).toBeTrue();
+					// Java arrays do not expose CFML array member functions on every engine.
+					var observations = [];
+					for ( var observation in states.toArray() ) {
+						observations.append( observation );
+					}
+					expect( observations.map( ( entry ) => entry.state ).find( "onCBQJobAttemptFinished" ) ).toBeLT(
+						observations.map( ( entry ) => entry.state ).find( "onCBQJobExecutionExited" )
+					);
+					var finished = observations.filter( ( entry ) => entry.state == "onCBQJobAttemptFinished" );
+					expect( finished.len() ).toBe( 1 );
+					expect( finished[ 1 ].status ).toBe( "deadline_exceeded" );
+					for ( var entry in observations ) {
+						expect( entry.id ).toBe( observations[ 1 ].id );
+					}
+				} finally {
+					executor.shutdownNow();
+				}
+			} );
+			it( "reports bulk publish rejection without replacing the provider failure", function() {
+				var states = [];
+				var dispatcher = createMock( "cbq.models.Jobs.Dispatcher" );
+				var diagnostics = [];
+				dispatcher.$property(
+					"log",
+					"variables",
+					{
+						warn : function( message, exception ) {
+							diagnostics.append( exception );
+							throw( type = "DiagnosticFailure" );
+						}
+					}
+				);
+				var connection = {
+					getDefaultQueue : function() {
+						return "synthetic";
+					},
+					pushMany : function() {
+						throw( type = "EnqueueFailure", message = "original enqueue failure" );
+					}
+				};
+				dispatcher.$property(
+					"config",
+					"variables",
+					{
+						getDefaultConnectionName : function() {
+							return "synthetic";
+						},
+						getConnection : function() {
+							return connection;
+						}
+					}
+				);
+				dispatcher.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : function( state, data ) {
+							states.append( state );
+							if ( state == "onCBQJobPublishException" ) {
+								throw( "ObserverFailure" );
+							}
+						}
+					}
+				);
+				var jobs = [
+					createMock( "cbq.models.Jobs.AbstractJob" ).init(),
+					createMock( "cbq.models.Jobs.AbstractJob" ).init()
+				];
+				expect( () => dispatcher.bulkDispatch( jobs = jobs, batchSize = 2 ) ).toThrow( "EnqueueFailure" );
+				expect( diagnostics.len() ).toBe( 2 );
+				expect( states ).toBe( [
+					"onCBQJobAdded",
+					"onCBQJobAdded",
+					"onCBQJobPublishException",
+					"onCBQJobPublishException"
+				] );
+			} );
+			it( "preserves successful publication when both its observer and diagnostic logger fail", function() {
+				var pushed = 0;
+				var dispatcher = createMock( "cbq.models.Jobs.Dispatcher" );
+				dispatcher.$property(
+					"log",
+					"variables",
+					{
+						warn : () => {
+							throw( type = "DiagnosticFailure" );
+						}
+					}
+				);
+				var connection = {
+					getDefaultQueue : () => "synthetic",
+					pushMany : function( entries ) {
+						pushed += entries.len();
+					}
+				};
+				dispatcher.$property(
+					"config",
+					"variables",
+					{
+						getDefaultConnectionName : () => "synthetic",
+						getConnection : () => connection
+					}
+				);
+				dispatcher.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : function( state, data ) {
+							if ( state == "onCBQJobPublished" ) {
+								throw( type = "ObserverFailure" );
+							}
+						}
+					}
+				);
+				var jobs = [
+					createMock( "cbq.models.Jobs.AbstractJob" ).init(),
+					createMock( "cbq.models.Jobs.AbstractJob" ).init()
+				];
+				expect( dispatcher.bulkDispatch( jobs = jobs, batchSize = 2 ) ).toBe( dispatcher );
+				expect( pushed ).toBe( 2 );
+			} );
+			it( "preserves the original synchronous job failure and worker exit despite failed observation diagnostics", function() {
+				var provider = createMock( "cbq.models.Providers.SyncProvider" );
+				var exited = false;
+				provider.$property(
+					"log",
+					"variables",
+					{
+						canDebug : () => false,
+						canError : () => false,
+						debug : () => {
+						},
+						warn : () => {
+							throw( type = "DiagnosticFailure" );
+						}
+					}
+				);
+				provider.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : function( state, data ) {
+							if ( state == "onCBQJobExecutionExited" ) {
+								exited = true;
+							}
+							if (
+								state == "onCBQJobExecutionStarted" || state == "onCBQJobAttemptFinished" || state == "onCBQJobExecutionExited"
+							) {
+								throw( type = "ObserverFailure" );
+							}
+						}
+					}
+				);
+				var job = createMock( "cbq.models.Jobs.AbstractJob" )
+					.init()
+					.setId( "synthetic" )
+					.setCurrentAttempt( 1 )
+					.setMaxAttempts( 1 )
+					.setMapping( "Synthetic" );
+				job.$( "handle" )
+					.$callback( () => {
+						throw( type = "OriginalJobFailure", message = "original synthetic job failure" );
+					} );
+				job.$( "isBatchJob", false );
+				var pool = createMock( "cbq.models.Workers.WorkerPool" ).init().setMaxAttempts( 1 );
+				expect( () => provider.marshalJob( job, pool ) ).toThrow( "OriginalJobFailure" );
+				expect( exited ).toBeTrue();
+			} );
+			it( "reports broken recovery diagnostics and terminal fallback without skipping the completion hook", function() {
+				var async = new coldbox.system.async.AsyncManager();
+				var executor = async.newExecutor(
+					"recovery-diagnostics-" & createUUID(),
+					"single",
+					1
+				);
+				var diagnostics = createObject( "java", "java.util.concurrent.ConcurrentLinkedQueue" ).init();
+				var forced = createObject( "java", "java.util.concurrent.atomic.AtomicBoolean" ).init( false );
+				var hookRan = createObject( "java", "java.util.concurrent.atomic.AtomicBoolean" ).init( false );
+				var provider = createMock( "cbq.models.Providers.AbstractQueueProvider" );
+				provider.$property( "async", "variables", async );
+				provider.$property(
+					"log",
+					"variables",
+					{
+						canDebug : () => false,
+						canError : () => true,
+						debug : () => {
+						},
+						error : () => {
+							throw( type = "DiagnosticFailure" );
+						},
+						warn : function( message, exception ) {
+							diagnostics.add( message );
+						}
+					}
+				);
+				provider.$property(
+					"interceptorService",
+					"variables",
+					{
+						announce : () => {
+						}
+					}
+				);
+				provider
+					.$( method = "forceFailJob", preserveReturnType = false )
+					.$callback( ( id, pool, job ) => {
+						forced.set( true );
+						throw( type = "TerminalPersistenceFailure" );
+					} );
+				var job = createMock( "cbq.models.Jobs.AbstractJob" )
+					.init()
+					.setId( "synthetic" )
+					.setMapping( "Synthetic" )
+					.setTimeout( 2 );
+				job.$( "handle" )
+					.$callback( () => {
+						throw( type = "OriginalJobFailure" );
+					} );
+				job.$( "isBatchJob", false );
+				var pool = createMock( "cbq.models.Workers.WorkerPool" )
+					.init()
+					.setExecutor( executor )
+					.setMaxAttempts( 1 );
+				try {
+					provider
+						.marshalJob(
+							job,
+							pool,
+							( completedJob, completedPool ) => hookRan.set( true )
+						)
+						.get( 2500 );
+					var messages = [];
+					for ( var message in diagnostics.toArray() ) {
+						messages.append( message );
+					}
+					expect( forced.get() ).toBeTrue();
+					expect( hookRan.get() ).toBeTrue();
+					expect( messages.len() ).toBe( 2 );
+					expect( messages[ 1 ] ).toInclude( "onException handler diagnostic" );
+					expect( messages[ 2 ] ).toInclude( "forceFailJob fallback" );
+				} finally {
+					executor.shutdownNow();
+				}
+			} );
+		} );
+	}
+
+}