@@ -0,0 +1,16 @@
+---
+'@objectstack/service-automation': patch
+---
+
+fix(service-automation): a connector definition saved, published or deleted on one replica now converges every replica
+
+Clause-②: no
+Metadata-protocol: none
+
+A connector instance definition changed through the per-item publish door re-materialized only on the replica that published it: that replica's `metadata:reloaded` hook ran the connector reconcile, but the hook is kernel-local, and the cluster's metadata-mutation lane (lane 2, `metadata.mutated`) delivered the change to the other replicas' mutation listeners, where the automation plugin acted on `flow` mutations only. Peers kept serving the old definition until a reload or a restart. A direct active save (`PUT /api/v1/meta/connector/:name`) raised no reload at all, so it did not re-materialize even the replica that wrote it.
+
+The automation plugin's metadata-mutation listener now queues the same soft reconcile `metadata:reloaded` runs for a `connector` mutation that is not a draft. No new cluster channel is added: a peer converges on the lane 2 replay it already receives, re-reading the shared store.
+
+- A publish on the writing replica reconciles from both of its signals; the second run finds the instance unchanged, so the provider is still called once per changed definition.
+- A direct active save re-materializes the writing replica and converges its peers. A deleted connector is torn down on every replica.
+- A draft save, and a mutation of any other type, runs no connector reconcile. Flow mutations are handled as before.
@@ -2,7 +2,8 @@
 
 /**
  * [#27977] Cross-replica reach of a connector DEFINITION published through the
- * per-item publish door — measured over the topology of
+ * per-item publish door — measured, and since #28002 pinned converging, over
+ * the topology of
  * `connector-credential-cluster.test.ts`: two kernels ("replicas") over ONE
  * shared database, each with its own materialized connectors, joined by a bus
  * that delivers every publish to every subscriber (publisher included).
@@ -22,13 +23,25 @@
  *  • lane 2 carries the mutation to B: B's protocol replays it into B's
  *    mutation listeners                                              -> GREEN
  *  • B's live instance follows the new definition with no reload on B
- *                                                                    -> RED
- *    (the automation plugin's mutation listener takes `flow` only, and
- *    `metadata:reloaded` is a kernel-local hook; pinned as `it.fails`, the
- *    fix is #28002)
+ *                                                                    -> GREEN
+ *    (measured RED by #27977: the automation plugin's mutation listener took
+ *    `flow` only, and `metadata:reloaded` is a kernel-local hook. #28002
+ *    widens that listener: a `connector` mutation, lane 2's replay included,
+ *    queues the reconcile `metadata:reloaded` runs)
  *  • control: a `metadata:reloaded` on B converges B from its OWN read of the
  *    shared store, so the gap is the trigger and not the store or the read
  *                                                                    -> GREEN
+ *
+ * [#28002] Added beside them:
+ *  • A publishes, so A reconciles from its own `metadata:reloaded` AND its own
+ *    mutation listener: the provider still runs once per changed definition
+ *                                                                    -> GREEN
+ *  • a direct active save (`PUT /api/v1/meta/connector/:name`, no publish
+ *    door) re-materializes the writing replica and converges the peer
+ *                                                                    -> GREEN
+ *  • a deleted connector is torn down on both replicas               -> GREEN
+ *  • controls: a draft save and a flow save run no connector reconcile; the
+ *    flow save still arms the flow                                   -> GREEN
  */
 import { mkdtempSync, rmSync } from 'node:fs';
 import { tmpdir } from 'node:os';
@@ -59,10 +72,24 @@ const definition = (endpoint: string) => ({
     providerConfig: { endpoint },
 });
 
-/** The protocol surface this suite drives: the doors `PUT /meta/:type/:name` and `POST …/publish` call. */
+/** A minimal start → end flow, saved active: the control's other metadata type. */
+const plainFlow = (name: string) => ({
+    name,
+    label: name,
+    type: 'autolaunched',
+    status: 'active',
+    nodes: [
+        { id: 'start', type: 'start', label: 'Start' },
+        { id: 'end', type: 'end', label: 'End' },
+    ],
+    edges: [{ id: 'e1', source: 'start', target: 'end' }],
+});
+
+/** The protocol surface this suite drives: the doors `PUT /meta/:type/:name`, `POST …/publish` and `DELETE` call. */
 interface Protocol {
     saveMetaItem(request: { type: string; name: string; item: unknown; mode?: 'draft' | 'publish' }): Promise<unknown>;
     publishMetaItem(request: { type: string; name: string }): Promise<unknown>;
+    deleteMetaItem(request: { type: string; name: string }): Promise<unknown>;
     onMetadataMutation(listener: (evt: { type: string; name: string; state?: string }) => void): () => void;
     attachMetadataMutationPubSub(pubsub: IPubSub, nodeId: string): () => void;
 }
@@ -105,14 +132,20 @@ function sharedBus(): IPubSub {
     };
 }
 
+/** Every connector reconcile, attributed to its plugin instance through `mock.contexts`. */
+const reconcileSpy = vi.spyOn(
+    AutomationServicePlugin.prototype as unknown as { reconcileDeclaredConnectors(...args: unknown[]): Promise<void> },
+    'reconcileDeclaredConnectors',
+);
+
 const kernels: ObjectKernel[] = [];
 const dirs: string[] = [];
 afterEach(async () => {
     for (const k of kernels.splice(0)) {
         if (k.getState() === 'running') await k.shutdown().catch(() => {});
     }
     for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true });
-    vi.restoreAllMocks();
+    reconcileSpy.mockClear();
 });
 
 /** The ONE database file both replicas open. */
@@ -122,14 +155,19 @@ function sharedDatabaseFile(): string {
     return join(dir, 'shared.sqlite');
 }
 
-/** One replica over the shared file, attached to `bus` the way the cluster bridge attaches it. */
-async function replica(file: string, nodeId: string, bus: IPubSub) {
+/**
+ * One replica over the shared file, attached to `bus` the way the cluster
+ * bridge attaches it. `lane2: false` leaves lane 2 off, the reach #27977
+ * measured a peer as having for this signal.
+ */
+async function replica(file: string, nodeId: string, bus: IPubSub, opts: { lane2?: boolean } = {}) {
     const provider = fakeProvider();
     let hooks: { trigger(event: string, payload?: unknown): Promise<void> } | undefined;
     const kernel = new ObjectKernel({ logger: { level: 'fatal' }, gracefulShutdown: false } as never);
     kernels.push(kernel);
     await kernel.use(new ObjectQLPlugin());
-    await kernel.use(new AutomationServicePlugin({ suspendedRunStore: 'memory', credentialResolver: () => undefined }));
+    const plugin = new AutomationServicePlugin({ suspendedRunStore: 'memory', credentialResolver: () => undefined });
+    await kernel.use(plugin);
     await kernel.use({
         name: `test.harness.${nodeId}`,
         type: 'standard' as const,
@@ -153,7 +191,7 @@ async function replica(file: string, nodeId: string, bus: IPubSub) {
     const protocol = kernel.getService<Protocol>('protocol');
     const automation = kernel.getService<AutomationEngine>('automation');
     // What `MetadataClusterBridgePlugin`'s lanes 2 and 4 do at `kernel:ready` on a cross-process driver.
-    protocol.attachMetadataMutationPubSub(bus, nodeId);
+    if (opts.lane2 !== false) protocol.attachMetadataMutationPubSub(bus, nodeId);
     automation.attachConnectorCredentialPubSub(bus, nodeId);
 
     const heard: string[] = [];
@@ -164,9 +202,16 @@ async function replica(file: string, nodeId: string, bus: IPubSub) {
         await protocol.saveMetaItem({ type: 'connector', name: NAME, item, mode: 'draft' });
         await protocol.publishMetaItem({ type: 'connector', name: NAME });
     };
+    /** The direct door: `PUT /api/v1/meta/connector/:name`, saved active. */
+    const save = (item: unknown) => protocol.saveMetaItem({ type: 'connector', name: NAME, item });
+    const saveDraft = (item: unknown) => protocol.saveMetaItem({ type: 'connector', name: NAME, item, mode: 'draft' });
+    const remove = () => protocol.deleteMetaItem({ type: 'connector', name: NAME });
+    const saveFlow = (name: string) => protocol.saveMetaItem({ type: 'flow', name, item: plainFlow(name) });
     const reload = () => hooks!.trigger('metadata:reloaded', { changed: [`connector/${NAME}`] });
     const live = () => automation.getRegisteredConnectors().includes(NAME);
-    return { provider, publish, reload, live, heard };
+    /** Connector reconciles this replica ran — the run `metadata:reloaded` queues. */
+    const reconciles = () => reconcileSpy.mock.contexts.filter((c) => c === plugin).length;
+    return { provider, publish, save, saveDraft, remove, saveFlow, reload, live, heard, reconciles, automation };
 }
 
 describe('[#27977] a connector definition published on one replica', () => {
@@ -185,8 +230,8 @@ describe('[#27977] a connector definition published on one replica', () => {
         await vi.waitFor(() => expect(b.heard).toContain(`connector/${NAME}`));
     });
 
-    // Measured by #27977; the fix is #28002. Promote to a plain `it` when #28002 lands.
-    it.fails('[#27977] converges the peer with no reload on the peer — known gap, filed as #28002', async () => {
+    // Measured RED by #27977 (pinned `it.fails`); converges since #28002.
+    it('[#28002] converges the peer with no reload on the peer', async () => {
         const file = sharedDatabaseFile();
         const bus = sharedBus();
         const a = await replica(file, 'node-a', bus);
@@ -198,17 +243,89 @@ describe('[#27977] a connector definition published on one replica', () => {
         await vi.waitFor(() => expect(b.provider.last()).toBe(V2), { timeout: 500 });
     });
 
-    it('control: a reload on the peer converges it from its own read of the shared store', async () => {
+    it('[#28002] the publishing replica materializes each changed definition once, though both its signals reconcile', async () => {
         const file = sharedDatabaseFile();
         const bus = sharedBus();
         const a = await replica(file, 'node-a', bus);
         const b = await replica(file, 'node-b', bus);
+        const before = a.reconciles();
+
+        await a.publish(definition(V1));
+        await a.publish(definition(V2));
+        await vi.waitFor(() => expect(b.provider.last()).toBe(V2), { timeout: 2000 });
+        for (let i = 0; i < 5; i++) await flush();
+
+        // Two reconciles per publish on A — its own `metadata:reloaded` and its
+        // own mutation listener (lane 2's loopback guard drops the bus echo) —
+        // and the second finds the signature unchanged.
+        expect(a.reconciles() - before).toBe(4);
+        expect(a.provider.endpoints).toEqual([V1, V2]);
+        expect(a.live()).toBe(true);
+    });
+
+    it('[#28002] a direct active save re-materializes the writing replica and converges the peer', async () => {
+        const file = sharedDatabaseFile();
+        const bus = sharedBus();
+        const a = await replica(file, 'node-a', bus);
+        const b = await replica(file, 'node-b', bus);
+
+        await a.save(definition(V1));
+        await vi.waitFor(() => expect(a.provider.last()).toBe(V1), { timeout: 2000 });
+        await vi.waitFor(() => expect(b.provider.last()).toBe(V1), { timeout: 2000 });
+
+        await a.save(definition(V2));
+        await vi.waitFor(() => expect(a.provider.last()).toBe(V2), { timeout: 2000 });
+        await vi.waitFor(() => expect(b.provider.last()).toBe(V2), { timeout: 2000 });
+        expect(a.live()).toBe(true);
+        expect(b.live()).toBe(true);
+    });
+
+    it('[#28002] a deleted connector is torn down on both replicas', async () => {
+        const file = sharedDatabaseFile();
+        const bus = sharedBus();
+        const a = await replica(file, 'node-a', bus);
+        const b = await replica(file, 'node-b', bus);
+
+        await a.publish(definition(V1));
+        await vi.waitFor(() => expect(b.live()).toBe(true), { timeout: 2000 });
+
+        await a.remove();
+        await vi.waitFor(() => expect(a.live()).toBe(false), { timeout: 2000 });
+        await vi.waitFor(() => expect(b.live()).toBe(false), { timeout: 2000 });
+    });
+
+    it('[#28002] control: a draft connector save and a flow save run no connector reconcile; the flow still arms', async () => {
+        const file = sharedDatabaseFile();
+        const bus = sharedBus();
+        const a = await replica(file, 'node-a', bus);
+        const b = await replica(file, 'node-b', bus);
+        await a.publish(definition(V1));
+        await vi.waitFor(() => expect(b.provider.last()).toBe(V1), { timeout: 2000 });
+        for (let i = 0; i < 5; i++) await flush();
+        const [beforeA, beforeB] = [a.reconciles(), b.reconciles()];
+
+        await a.saveDraft(definition(V2));
+        await a.saveFlow('flow_28002');
+        await vi.waitFor(async () => expect(await a.automation.getFlow('flow_28002')).not.toBeNull(), { timeout: 2000 });
+        await vi.waitFor(async () => expect(await b.automation.getFlow('flow_28002')).not.toBeNull(), { timeout: 2000 });
+        for (let i = 0; i < 5; i++) await flush();
+
+        expect([a.reconciles() - beforeA, b.reconciles() - beforeB]).toEqual([0, 0]);
+        expect([a.provider.endpoints, b.provider.endpoints]).toEqual([[V1], [V1]]);
+    });
+
+    it('control: a reload on the peer converges it from its own read of the shared store', async () => {
+        const file = sharedDatabaseFile();
+        const bus = sharedBus();
+        const a = await replica(file, 'node-a', bus);
+        // Lane 2 off on the peer: the reload is then its only trigger.
+        const b = await replica(file, 'node-b', bus, { lane2: false });
 
         await a.publish(definition(V1));
         await a.publish(definition(V2));
-        await vi.waitFor(() => expect(b.heard).toContain(`connector/${NAME}`));
         for (let i = 0; i < 5; i++) await flush();
-        // The measured state before the reload: the peer never materialized the instance.
+        // The state before the reload: the peer never materialized the instance.
+        expect(b.heard).toEqual([]);
         expect(b.live()).toBe(false);
         expect(b.provider.endpoints).toEqual([]);
 
 
@@ -842,7 +842,13 @@ export class AutomationServicePlugin implements Plugin {
      * registration per publish, not two.
      */
     private mutationArmedFlows = new Map<string, string>();
-    /** [#21725] The `onMetadataMutation` subscription; dropped at destroy. */
+    /**
+     * [#28002] A connector reconcile the metadata mutation signal queued and
+     * that has not started yet: every connector mutation reported before it
+     * starts is answered by that one run, which re-reads the whole declaration.
+     */
+    private connectorMutationReconcilePending = false;
+    /** [#21725] The `onMetadataMutation` subscription (flows, and connectors since #28002); dropped at destroy. */
     private unsubscribeFlowMutations?: () => void;
 
     constructor(options: AutomationServicePluginOptions = {}) {
@@ -3642,6 +3648,9 @@ export class AutomationServicePlugin implements Plugin {
      * run on the flow sync queue ({@link syncMutatedFlows}). The save's own
      * answer therefore precedes the arming by that one read.
      *
+     * [#28002] A `connector` mutation queues a connector reconcile instead
+     * ({@link queueConnectorMutationReconcile}); every other type is ignored.
+     *
      * Resolved at `start()`, once every plugin has inited, so a protocol that
      * registers after this plugin is seen.
      */
@@ -3668,15 +3677,50 @@ export class AutomationServicePlugin implements Plugin {
             return;
         }
         this.unsubscribeFlowMutations = protocol.onMetadataMutation((evt) => {
-            if (evt?.type !== 'flow' || evt.state === 'draft') return;
+            if (!evt || evt.state === 'draft') return;
             if (typeof evt.name !== 'string' || evt.name === '') return;
+            if (evt.type === 'connector') {
+                this.queueConnectorMutationReconcile(ctx);
+                return;
+            }
+            if (evt.type !== 'flow') return;
             const idle = this.pendingFlowMutations.size === 0;
             this.pendingFlowMutations.add(evt.name);
             // One queued sync drains every name reported before it starts.
             if (idle) void this.enqueueFlowSync(() => this.syncMutatedFlows(ctx)).catch(() => undefined);
         });
     }
 
+    /**
+     * [#28002] Reconcile this replica's connectors after a connector was saved
+     * active, published or deleted — the reconcile `metadata:reloaded` runs
+     * ({@link materializeDeclaredConnectors}, soft mode), on the reconcile
+     * queue. The mutation signal is what reaches a PEER: lane 2 of the cluster
+     * family replays a write made on another replica into this protocol's
+     * listeners, while `metadata:reloaded` is a kernel-local hook. It also
+     * covers a direct active save (`PUT /api/v1/meta/connector/:name`), which
+     * raises no reload anywhere.
+     *
+     * A publish on this replica raises both signals; the second reconcile
+     * finds every signature unchanged and leaves the instances alone.
+     * Coalesced: a reconcile queued and not yet started answers every
+     * connector mutation reported before it starts.
+     */
+    private queueConnectorMutationReconcile(ctx: PluginContext): void {
+        if (this.connectorMutationReconcilePending) return;
+        this.connectorMutationReconcilePending = true;
+        void this.onReconcileQueue(async () => {
+            this.connectorMutationReconcilePending = false;
+            await this.reconcileDeclaredConnectors(ctx, { fatal: false });
+        }).catch((err) => {
+            ctx.logger.warn(
+                '[Automation] reconciling the connectors after a connector metadata change failed; '
+                    + 'the next metadata reload or restart reconciles them.',
+                describeThrownForLog(err),
+            );
+        });
+    }
+
     /**
      * [#21725] Make the engine's registration of each reported flow follow its
      * stored state — what a restart would arm, without the restart: