feat(memory-core): add harness presence address dispatch (#12422) (#12554)

Co-authored-by: tobiu <tobiasuhlig78@gmail.com>

Author: neo-gpt

ai/daemons/bridge/daemon.mjs

@@ -46,7 +46,9 @@ import {
     getGraphLogEntries,
     getNodesData,
     getEdgesData,
-    getDbNode
+    getDbNode,
+    getActiveHarnessPresence,
+    isHarnessPresenceFresh
 } from './queries.mjs';
 import {applyHarnessMetadataDefaults} from '../../scripts/lifecycle/harnessRouting.mjs';
 import {getDefaultInstancePid, getInstancePid} from './instanceResolver.mjs';
@@ -676,6 +678,7 @@ let deliveryPromise = Promise.resolve();
  */
 async function deliverDigest(subscription, digest) {
     const meta = subscription.properties?.harnessTargetMetadata || {};
+    const {instanceAddress, addressType} = resolveInstanceAddress(meta);
     // Fall back to osascript on macOS by default, tmux otherwise
     const defaultAdapter = process.platform === 'darwin' ? 'osascript' : 'tmux';
     const adapter = meta.adapter || defaultAdapter;
@@ -684,8 +687,19 @@ async function deliverDigest(subscription, digest) {
     deliveryPromise = deliveryPromise.then(async () => {
 
     try {
+        if (addressType && instanceAddress && !assertFreshTargetPresence(subscription, {addressType})) {
+            return;
+        }
+
+        if (addressType === 'webhookUrl') {
+            await deliverViaWebhookUrl(subscription, digest, instanceAddress);
+            return;
+        }
+
         if (adapter === 'tmux') {
-            const tmuxSession = meta.tmuxSession || process.env.TMUX_SESSION || 'neo-agent';
+            const tmuxSession = addressType === 'tmuxSession' && instanceAddress
+                ? instanceAddress
+                : meta.tmuxSession || process.env.TMUX_SESSION || 'neo-agent';
             await spawnAsync('tmux', ['send-keys', '-t', tmuxSession, digest, 'C-m']);
             writeLog('INFO', `[Bridge Daemon] Delivered ${subscription.id} via tmux to session ${tmuxSession}`);
         } else if (adapter === 'osascript') {
@@ -722,11 +736,10 @@ async function deliverDigest(subscription, digest) {
                 return;
             }
 
-            // Instance-addressable wake — LOCAL deployment only. When the subscription carries a
-            // userDataDir, two same-bundle GUI harnesses may be running; resolve the intended
-            // instance's pid and raise THAT process — never the ambiguous app-activate / frontmost
-            // guess. Fail closed (skip the wake) if the instance cannot be located, so a targeted
-            // wake never lands in the wrong one.
+            // Instance-addressable wake — LOCAL deployment only. When the subscription carries an
+            // addressType, route through that address instead of falling back to ambiguous
+            // app-activate/frontmost guessing. Fail closed (skip the wake) if the instance cannot
+            // be located, so a targeted wake never lands in the wrong one.
             //
             // Instance addressing is a local-only primitive: this daemon delivers desktop-harness
             // wakes via osascript/tmux, which a headless cloud deployment has no GUI harness to
@@ -736,7 +749,24 @@ async function deliverDigest(subscription, digest) {
             // app-activate — a targeted wake must never silently degrade to an untargeted one. Uses
             // the canonical AiConfig.orchestrator.deploymentMode signal.
             let instancePid = null;
-            if (meta.userDataDir) {
+            if (addressType === 'pid') {
+                if (AiConfig.orchestrator.deploymentMode === 'cloud') {
+                    writeLog('ERROR',
+                        `[Bridge Daemon] Instance wake refused for ${subscription.id}: ` +
+                        `pid targeting requires local deployment (deploymentMode='cloud'). ` +
+                        `Failing closed — instance-addressed GUI wakes are local-only (ADR 0014).`
+                    );
+                    return;
+                }
+                instancePid = normalizePid(instanceAddress);
+                if (!instancePid) {
+                    writeLog('ERROR',
+                        `[Bridge Daemon] Instance wake refused for ${subscription.id}: ` +
+                        `invalid pid instanceAddress='${instanceAddress}'. Failing closed.`
+                    );
+                    return;
+                }
+            } else if (addressType === 'userDataDir' && instanceAddress) {
                 if (AiConfig.orchestrator.deploymentMode === 'cloud') {
                     writeLog('ERROR',
                         `[Bridge Daemon] Instance wake refused for ${subscription.id}: ` +
@@ -745,11 +775,11 @@ async function deliverDigest(subscription, digest) {
                     );
                     return;
                 }
-                instancePid = await getInstancePid({userDataDir: meta.userDataDir});
+                instancePid = await getInstancePid({userDataDir: instanceAddress});
                 if (!instancePid) {
                     writeLog('ERROR',
                         `[Bridge Daemon] Instance wake refused for ${subscription.id}: ` +
-                        `no running process found for userDataDir='${meta.userDataDir}'. ` +
+                        `no running process found for userDataDir='${instanceAddress}'. ` +
                         `Failing closed to avoid misrouting to another ${appName} instance.`
                     );
                     return;
@@ -945,6 +975,94 @@ async function deliverDigest(subscription, digest) {
     return deliveryPromise;
 }
 
+/**
+ * @summary Resolves generic instance-address metadata with legacy field compatibility.
+ * @param {Object} meta Subscription harnessTargetMetadata.
+ * @returns {{instanceAddress:String|null,addressType:String|null}}
+ */
+function resolveInstanceAddress(meta = {}) {
+    const addressType = meta.addressType
+        || (meta.userDataDir ? 'userDataDir' : null);
+
+    const instanceAddress = meta.instanceAddress
+        || (addressType === 'userDataDir' ? meta.userDataDir : null);
+
+    return {
+        instanceAddress: instanceAddress || null,
+        addressType    : addressType || null
+    };
+}
+
+/**
+ * @summary Requires fresh HarnessPresence before immediate address-specific dispatch.
+ * @param {Object} subscription WAKE_SUBSCRIPTION node.
+ * @param {Object} address Resolved address tuple.
+ * @returns {Boolean}
+ */
+function assertFreshTargetPresence(subscription, {addressType}) {
+    const presence = getActiveHarnessPresence(db, {
+        subscriptionId: subscription.id,
+        agentIdentity : subscription.properties?.agentIdentity
+    });
+
+    if (isHarnessPresenceFresh(presence)) return true;
+
+    writeLog('WARN',
+        `[Bridge Daemon] Targeted wake refused for ${subscription.id}: ` +
+        `addressType='${addressType}' requires fresh HarnessPresence. ` +
+        `Failing closed; recipient will pick up the unread event on next turn.`
+    );
+    return false;
+}
+
+/**
+ * @summary Normalizes a pid string for direct-pid address dispatch.
+ * @param {*} pid Process id candidate.
+ * @returns {Number|null}
+ */
+function normalizePid(pid) {
+    const numericPid = Number(pid);
+
+    return Number.isInteger(numericPid) && numericPid > 0 ? numericPid : null;
+}
+
+/**
+ * @summary Posts a wake digest to a bridge-dispatchable webhook address.
+ * @param {Object} subscription WAKE_SUBSCRIPTION node.
+ * @param {String} digest Wake digest body.
+ * @param {String} webhookUrl Target webhook URL.
+ * @returns {Promise<void>}
+ */
+async function deliverViaWebhookUrl(subscription, digest, webhookUrl) {
+    let url;
+    try {
+        url = new URL(webhookUrl);
+    } catch (error) {
+        writeLog('ERROR',
+            `[Bridge Daemon] Webhook wake refused for ${subscription.id}: ` +
+            `invalid webhookUrl instanceAddress. Failing closed.`
+        );
+        return;
+    }
+
+    const response = await fetch(url, {
+        method : 'POST',
+        headers: {
+            'content-type': 'application/json'
+        },
+        body: JSON.stringify({
+            subscriptionId: subscription.id,
+            digest
+        })
+    });
+
+    if (!response.ok) {
+        throw new Error(`webhookUrl POST failed with HTTP ${response.status}`);
+    }
+
+    writeLog('INFO', `[Bridge Daemon] Delivered ${subscription.id} via webhookUrl POST`);
+}
+
 // Start loop
 async function main() {
     await enforceSingleton();

ai/daemons/bridge/queries.mjs

@@ -38,6 +38,85 @@ export function getActiveShapeCSubscriptions(db) {
     return collapseDuplicateShapeCRoutes(stmt.all().map(row => JSON.parse(row.data)));
 }
 
+export const HARNESS_PRESENCE_STALE_AFTER_MS = 5 * 60 * 1000;
+
+/**
+ * @summary Loads the newest active HarnessPresence row for a bridge-daemon subscription.
+ *
+ * Presence is a volatile overlay keyed by subscription and identity. The bridge daemon consumes it
+ * only as a freshness guard for targeted delivery; missing or stale presence must fail closed for
+ * address-specific dispatch instead of falling through to the legacy untargeted activate path.
+ *
+ * @param {Database} db SQLite database handle.
+ * @param {Object} opts
+ * @param {String} opts.subscriptionId WAKE_SUBSCRIPTION id.
+ * @param {String} opts.agentIdentity AgentIdentity node id.
+ * @returns {Object|null} Parsed HARNESS_PRESENCE node.
+ */
+export function getActiveHarnessPresence(db, {subscriptionId, agentIdentity} = {}) {
+    if (!subscriptionId && !agentIdentity) return null;
+
+    if (subscriptionId) {
+        return getNewestHarnessPresence(db, {
+            where : "json_extract(data, '$.properties.subscriptionId') = ?",
+            params: [subscriptionId]
+        });
+    }
+
+    if (!agentIdentity) return null;
+
+    return getNewestHarnessPresence(db, {
+        where : "json_extract(data, '$.properties.agentIdentity') = ?",
+        params: [agentIdentity]
+    });
+}
+
+function getNewestHarnessPresence(db, {where, params}) {
+    const rows = db.prepare(`
+        SELECT data FROM Nodes
+        WHERE json_extract(data, '$.label') = 'HARNESS_PRESENCE'
+          AND COALESCE(json_extract(data, '$.properties.status'), 'active') = 'active'
+          AND ${where}
+        ORDER BY COALESCE(
+            json_extract(data, '$.properties.lastSeenAt'),
+            json_extract(data, '$.properties.updatedAt'),
+            ''
+        ) DESC
+        LIMIT 1
+    `).all(...params);
+
+    for (const row of rows) {
+        try {
+            return JSON.parse(row.data);
+        } catch (error) {
+            return null;
+        }
+    }
+
+    return null;
+}
+
+/**
+ * @summary Checks whether a HarnessPresence row is fresh enough for immediate targeted delivery.
+ * @param {Object|null} presence Parsed HARNESS_PRESENCE node.
+ * @param {Object} [opts]
+ * @param {Number} [opts.now=Date.now()] Current timestamp in ms.
+ * @param {Number} [opts.staleAfterMs=HARNESS_PRESENCE_STALE_AFTER_MS] Staleness threshold.
+ * @returns {Boolean}
+ */
+export function isHarnessPresenceFresh(presence, {
+    now = Date.now(),
+    staleAfterMs = HARNESS_PRESENCE_STALE_AFTER_MS
+} = {}) {
+    const props = presence?.properties || {};
+    if ((props.status || 'active') !== 'active') return false;
+
+    const lastSeenAt = props.lastSeenAt ? new Date(props.lastSeenAt).getTime() : NaN;
+    if (!Number.isFinite(lastSeenAt)) return false;
+
+    return now - lastSeenAt <= staleAfterMs;
+}
+
 /**
  * @summary Collapses duplicate active Shape C wake routes before bridge-daemon dispatch.
  *

ai/mcp/server/memory-core/openapi.yaml

@@ -1351,14 +1351,16 @@ paths:
                   description: Required for subscribe; specifies how wake events are delivered.
                 harnessTargetMetadata:
                   type: object
-                  description: Channel-specific metadata. appName is required so fresh agents always provide the bridge-daemon/osascript routing target; url is required for a2a-webhook; coalesceWindow is an optional override. userDataDir (local bridge-daemon only) addresses one specific same-bundle GUI instance when two harnesses of the same app run in parallel — the daemon resolves that instance's pid from its --user-data-dir and raises only that process.
+                  description: Channel-specific metadata. appName is required so fresh agents always provide the bridge-daemon/osascript routing target; url is required for a2a-webhook; coalesceWindow is an optional override. instanceAddress + addressType (local bridge-daemon only) target a specific receiver instance; userDataDir is a legacy compatibility field for existing subscriptions.
                   required: [appName]
                   properties:
                     url:              {type: string}
                     coalesceWindow:   {type: integer, minimum: 0, maximum: 300}
                     daemonSocketPath: {type: string}
                     adapter:          {type: string, enum: [osascript, tmux]}
                     appName:          {type: string}
+                    instanceAddress:  {type: string, nullable: true}
+                    addressType:      {type: string, enum: [userDataDir, pid, tmuxSession, webhookUrl], nullable: true}
                     userDataDir:      {type: string, nullable: true}
                     tabShortcut:      {type: string, nullable: true}
                     focusSeedKey:     {type: string, nullable: true}

ai/mcp/server/shared/services/BootEnvelopeResolver.mjs

@@ -33,12 +33,11 @@ import Base from '../../../../../src/core/Base.mjs';
  *
  * ## Dispatch coverage
  *
- * The bridge daemon currently dispatches the `userDataDir` address kind. The `pid`, `tmuxSession`,
- * and `webhookUrl` kinds are recognized envelope values reserved for the forthcoming generic
- * address-type dispatch; until that lands they fail closed here rather than produce a subscription
- * the daemon would misroute. For `userDataDir`, the address is also mirrored onto the legacy
- * `userDataDir` metadata field the daemon reads today; that mirror is a transitional bridge and is
- * retired once dispatch reads the generic `addressType` directly.
+ * The bridge daemon dispatches the generic `{instanceAddress, addressType}` pair directly:
+ * `userDataDir` resolves to a same-bundle GUI process, `pid` targets that process directly,
+ * `tmuxSession` sends to tmux, and `webhookUrl` posts the wake digest. The earlier transitional
+ * `userDataDir` mirror is retired here; legacy subscriptions that already carry that field remain
+ * a bridge-daemon compatibility read concern, not a boot-envelope output.
  *
  * Pure resolution is intentionally side-effect-free (reads `process.env` only) so the mapping is
  * unit-testable with injected environments.
@@ -65,16 +64,15 @@ class BootEnvelopeResolver extends Base {
 
     /**
      * @member {String[]} validAddressTypes
-     * The recognized instance-address kinds. `userDataDir` is dispatched today; the remainder are
-     * reserved for generic address-type dispatch and currently fail closed.
+     * The recognized instance-address kinds.
      */
     validAddressTypes = ['userDataDir', 'pid', 'tmuxSession', 'webhookUrl']
 
     /**
      * @member {String[]} dispatchableAddressTypes
      * The subset of {@link validAddressTypes} the bridge daemon can route today.
      */
-    dispatchableAddressTypes = ['userDataDir']
+    dispatchableAddressTypes = ['userDataDir', 'pid', 'tmuxSession', 'webhookUrl']
 
     /**
      * Resolves the boot instance-address envelope into wake-subscription `overrideMetadata`.
@@ -123,16 +121,7 @@ class BootEnvelopeResolver extends Base {
             );
         }
 
-        const overrideMetadata = {instanceAddress, addressType};
-
-        // Transitional daemon-compat: the live bridge daemon resolves the target instance from the
-        // `userDataDir` metadata field. Mirror the address there so wake routing is functional now;
-        // the mirror is retired once dispatch reads `addressType` directly.
-        if (addressType === 'userDataDir') {
-            overrideMetadata.userDataDir = instanceAddress;
-        }
-
-        return overrideMetadata;
+        return {instanceAddress, addressType};
     }
 }