test(control-plane): count indented read writes

TASK-565fa463-c9b0-49e6-a902-9f12a49b6615

docs/TEST_INTEGRITY.md

@@ -2616,7 +2616,7 @@ test_integrity_note:
     - "requestedLimit, returnedCount, complete, truncated and pendingUnreadCount"
     - "Zero INSERT/UPDATE/DELETE/REPLACE or equivalent write executions at the fixture connection boundary"
     - "Equal message ordering, acknowledgements, audit, identity, task, ownership and all present durable-table state before and after each observation"
-    - "Original implementation passes the full invariant; a test-local post-read audit INSERT mutant fails it and is restored deterministically"
+    - "Original implementation passes the full invariant; helper-based and direct leading-whitespace multiline audit INSERT mutants both fail the fixture-connection write counter and are restored deterministically"
   determinism_controls:
     - "Disposable Git and XDG fixtures with deterministic timestamps and message IDs"
     - "No provider, host, network, live-project database or runtime activation"
@@ -2626,6 +2626,7 @@ test_integrity_note:
       - "Returning independently sampled page and count metadata"
       - "WAL-only audit or acknowledgement writes hidden by unchanged main-database bytes"
       - "A source mutation that adds a post-read write while preserving message output"
+      - "A leading-whitespace multiline write bypassing connection-boundary DML classification"
       - "Silently coercing invalid selection or limit inputs"
   relaxation:
     did_relax_any_assertion: false

test/fleet/control-plane.test.mjs

@@ -43,7 +43,7 @@ function deterministicOptions(fixture) {
 }
 
 function writeCapableSql(sql) {
-  return /(?:^|[);]\s*)(?:INSERT|UPDATE|DELETE|REPLACE|CREATE|ALTER|DROP|VACUUM)\b/i.test(sql);
+  return /(?:^\s*|[);]\s*)(?:INSERT|UPDATE|DELETE|REPLACE|CREATE|ALTER|DROP|VACUUM)\b/i.test(sql);
 }
 
 function quoteIdentifier(identifier) {
@@ -221,13 +221,38 @@ test('SCN-control-plane-unread-observation-metadata: a bound inbox observation i
     assert.throws(
       () => observeWithoutDurableWrite(control, fixture,
         { recipient: fixture.worker, selection: 'unread', limit: 1 }),
-      (error) => /DML write execution/.test(error.message),
+      (error) => /DML write execution/.test(error.message)
+        && /INSERT INTO audit_events/.test(error.message),
     );
   } finally {
     control.observeMessages = originalObserveMessages;
   }
   assert.equal(control.observeMessages, originalObserveMessages,
     'the mutation wrapper must be restored after its required invariant failure');
+  control.observeMessages = function directMultilineAuditedReadMutant(input) {
+    const observed = originalObserveMessages.call(this, input);
+    this.database.prepare(`
+      INSERT INTO audit_events (id, project_id, actor_id, operation, entity_type, created_at)
+      VALUES (?, ?, ?, ?, ?, ?)
+    `).run(
+      'direct-multiline-diagnostic', this.projectId, input.recipient,
+      'diagnostic.direct-read-write', 'message-observation', '2026-10-01T12:00:00Z',
+    );
+    return observed;
+  };
+  try {
+    assert.throws(
+      () => observeWithoutDurableWrite(control, fixture,
+        { recipient: fixture.worker, selection: 'unread', limit: 1 }),
+      (error) => /executed 1 DML write execution/.test(error.message)
+        && /INSERT INTO audit_events/.test(error.message),
+      'a leading-whitespace direct multiline INSERT must be counted at the fixture connection boundary',
+    );
+  } finally {
+    control.observeMessages = originalObserveMessages;
+  }
+  assert.equal(control.observeMessages, originalObserveMessages,
+    'the direct multiline mutation wrapper must be restored after its required invariant failure');
   assert.deepEqual(observeWithoutDurableWrite(control, fixture,
     { recipient: fixture.worker, selection: 'unread', limit: 1 }).messages, [],
   'the restored original observation still passes the complete no-write invariant');