fix: make bridge rpc text sends idempotent

CHANGELOG.md

@@ -22,6 +22,9 @@
 ### Advanced IMCore
 - feat: add bridge-backed Apple URL preview sends through `send-rich --url` and `send.rich`, with an eight-second out-of-process preparation deadline, capped image decode/staging, and metadata-only fallback (#165, thanks @omarshahine).
 
+### JSON-RPC
+- fix: atomically claim bridge RPC inbox files before dispatch so multiple injected consumers cannot deliver one logical send twice (#158).
+
 ## 0.12.3 - 2026-07-06
 
 ### Native Polls

Sources/IMsgHelper/IMsgInjected.m

@@ -22,6 +22,7 @@
 #import <pwd.h>
 #import <errno.h>
 #import <stdio.h>
+#import <signal.h>
 #import <sys/stat.h>
 #import <dlfcn.h>
 
@@ -134,11 +135,12 @@ - (instancetype)initWithCoder:(NSCoder *)coder {
 
 static NSTimer *fileWatchTimer = nil;
 static NSTimer *rpcInboxTimer = nil;
-static NSMutableSet *processedRpcIds = nil;
+static BOOL bridgeDidBootstrap = NO;
 static os_unfair_lock eventsLock = OS_UNFAIR_LOCK_INIT;
 static int lockFd = -1;
 
 static const NSUInteger kEventsRotateBytes = 1 * 1024 * 1024;
+static const NSTimeInterval kV2ClaimMaxAge = 10 * 60;
 
 static void initFilePaths(void) {
     if (kCommandFile == nil) {
@@ -155,9 +157,6 @@ static void initFilePaths(void) {
         kEventsRotated = [containerPath stringByAppendingPathComponent:@".imsg-events.jsonl.1"];
         kDebugLogFile = [containerPath stringByAppendingPathComponent:@".imsg-bridge.log"];
     }
-    if (processedRpcIds == nil) {
-        processedRpcIds = [NSMutableSet set];
-    }
 }
 
 /// Append a line to `.imsg-bridge.log` inside the Messages container. NSLog
@@ -5509,26 +5508,39 @@ static void registerEventObservers(void) {
 
 #pragma mark - v2 Inbox Watcher
 
-/// Process a single inbox file end-to-end: read, dispatch, write outbox,
-/// remove inbox. Skips re-processed ids via processedRpcIds.
+/// Process a single inbox file end-to-end: claim, read, dispatch, write outbox,
+/// remove claim. The same-directory rename is the cross-process ownership
+/// boundary: only one injected process can remove the finalized request name.
 static void processV2InboxFile(NSString *uuid) {
     @autoreleasepool {
-        if ([processedRpcIds containsObject:uuid]) {
-            return;
-        }
-        [processedRpcIds addObject:uuid];
-
         NSString *inPath = [kRpcInDir stringByAppendingPathComponent:
             [uuid stringByAppendingPathExtension:@"json"]];
+        NSString *claimName = [NSString stringWithFormat:@"%@.processing.%d", uuid, getpid()];
+        NSString *claimPath = [kRpcInDir stringByAppendingPathComponent:claimName];
         NSString *outPath = [kRpcOutDir stringByAppendingPathComponent:
             [uuid stringByAppendingPathExtension:@"json"]];
 
+        if (rename(inPath.UTF8String, claimPath.UTF8String) != 0) {
+            int claimErrno = errno;
+            // ENOENT is the expected race loser: another injected process
+            // already claimed this request. Other failures leave the finalized
+            // file in place for a later scan instead of risking duplicate work.
+            if (claimErrno != ENOENT) {
+                NSLog(@"[imsg-bridge v2] Could not claim %@ (errno=%d)", inPath, claimErrno);
+                debugLog(@"v2 claim failed id=%@ pid=%d errno=%d",
+                         uuid, getpid(), claimErrno);
+            }
+            return;
+        }
+        debugLog(@"v2 claimed id=%@ pid=%d process=%@",
+                 uuid, getpid(), [[NSProcessInfo processInfo] processName]);
+
         NSError *err = nil;
-        NSData *body = [NSData dataWithContentsOfFile:inPath options:0 error:&err];
+        NSData *body = [NSData dataWithContentsOfFile:claimPath options:0 error:&err];
         if (!body || err) {
-            NSLog(@"[imsg-bridge v2] Could not read %@: %@", inPath, err);
+            NSLog(@"[imsg-bridge v2] Could not read %@: %@", claimPath, err);
             // Remove malformed file so we don't retry forever.
-            [[NSFileManager defaultManager] removeItemAtPath:inPath error:nil];
+            [[NSFileManager defaultManager] removeItemAtPath:claimPath error:nil];
             return;
         }
 
@@ -5552,12 +5564,49 @@ static void processV2InboxFile(NSString *uuid) {
             rename(tmp.UTF8String, outPath.UTF8String);
         }
 
-        // Drop the inbox request — we're done with it.
-        [[NSFileManager defaultManager] removeItemAtPath:inPath error:nil];
+        // Drop the claimed request — we're done with it. If the process dies
+        // after claiming, a later inbox scan removes the orphan without
+        // replaying a potentially delivered side effect.
+        [[NSFileManager defaultManager] removeItemAtPath:claimPath error:nil];
+    }
+}
+
+static pid_t v2ClaimOwnerPID(NSString *name) {
+    NSRange marker = [name rangeOfString:@".processing." options:NSBackwardsSearch];
+    if (marker.location == NSNotFound) return 0;
+
+    NSString *pidString = [name substringFromIndex:NSMaxRange(marker)];
+    NSScanner *scanner = [NSScanner scannerWithString:pidString];
+    int value = 0;
+    if (![scanner scanInt:&value] || !scanner.isAtEnd || value <= 0) return 0;
+    return (pid_t)value;
+}
+
+/// Claimed requests are never replayed: the handler may have dispatched its
+/// side effect before dying. Delete claims owned by dead processes, claims
+/// left by a recycled current PID, and old claims whose PID was reused by an
+/// unrelated live process.
+static void cleanupOrphanedV2Claims(NSArray *entries) {
+    NSFileManager *fm = [NSFileManager defaultManager];
+    for (NSString *name in entries) {
+        pid_t ownerPID = v2ClaimOwnerPID(name);
+        if (ownerPID <= 0) continue;
 
-        // Cap the dedupe set to prevent unbounded growth on long-lived dylibs.
-        if (processedRpcIds.count > 1024) {
-            [processedRpcIds removeAllObjects];
+        NSString *path = [kRpcInDir stringByAppendingPathComponent:name];
+        BOOL shouldRemove = ownerPID == getpid();
+        if (!shouldRemove && kill(ownerPID, 0) != 0 && errno == ESRCH) {
+            shouldRemove = YES;
+        }
+        if (!shouldRemove) {
+            NSDictionary *attrs = [fm attributesOfItemAtPath:path error:nil];
+            NSDate *modified = attrs[NSFileModificationDate];
+            shouldRemove =
+                modified && [[NSDate date] timeIntervalSinceDate:modified] > kV2ClaimMaxAge;
+        }
+        if (shouldRemove) {
+            [fm removeItemAtPath:path error:nil];
+            debugLog(@"v2 removed orphan claim=%@ owner_pid=%d",
+                     name, (int)ownerPID);
         }
     }
 }
@@ -5568,6 +5617,7 @@ static void scanV2Inbox(void) {
         NSArray *entries = [[NSFileManager defaultManager]
             contentsOfDirectoryAtPath:kRpcInDir error:&err];
         if (!entries) return;
+        cleanupOrphanedV2Claims(entries);
         for (NSString *name in entries) {
             // Only consume finalized .json files; skip in-flight .tmp.
             if (![name hasSuffix:@".json"]) continue;
@@ -5618,6 +5668,13 @@ static void bridgeBootstrap(void) {
     static dispatch_once_t once;
     dispatch_once(&once, ^{
         @autoreleasepool {
+            NSString *bundleIdentifier = [[NSBundle mainBundle] bundleIdentifier];
+            if (![bundleIdentifier isEqualToString:@"com.apple.MobileSMS"]) {
+                NSLog(@"[imsg-bridge] Ignoring injected process with bundle %@",
+                      bundleIdentifier ?: @"(none)");
+                return;
+            }
+            bridgeDidBootstrap = YES;
             initFilePaths();
             NSLog(@"[imsg-bridge] Dylib injected into %@",
                   [[NSProcessInfo processInfo] processName]);
@@ -5684,6 +5741,8 @@ static void injectedInit(void) {
 
 __attribute__((destructor))
 static void injectedCleanup(void) {
+    if (!bridgeDidBootstrap) return;
+
     NSLog(@"[imsg-bridge] Cleaning up...");
 
     if (fileWatchTimer) {

Sources/imsg/Commands/PollCommand.swift

@@ -167,6 +167,7 @@ enum PollCommand {
     storeFactory: @escaping (String) throws -> MessageStore,
     invokeBridge: @escaping (BridgeAction, [String: Any]) async throws -> [String: Any]
   ) async throws {
+    try validateOptionSelector(values)
     let chat = try resolveChatGUID(values: values, storeFactory: storeFactory)
     guard let pollGuid = values.option("poll")?.trimmingCharacters(in: .whitespacesAndNewlines),
       !pollGuid.isEmpty
@@ -225,11 +226,7 @@ enum PollCommand {
   }
 
   /// Resolve exactly one option selector against the poll's stable options.
-  private static func resolveOptionID(
-    values: ParsedValues,
-    pollGuid: String,
-    store: MessageStore
-  ) throws -> (id: String, text: String?) {
+  private static func validateOptionSelector(_ values: ParsedValues) throws {
     let directID = values.option("optionID")?.trimmingCharacters(in: .whitespacesAndNewlines)
     let indexValue = values.optionInt64("optionIndex")
     let textValue = values.option("option")?.trimmingCharacters(in: .whitespacesAndNewlines)
@@ -241,6 +238,17 @@ enum PollCommand {
       throw ParsedValuesError.invalidOption(
         "choose exactly one of --option-id, --option-index, or --option")
     }
+  }
+
+  private static func resolveOptionID(
+    values: ParsedValues,
+    pollGuid: String,
+    store: MessageStore
+  ) throws -> (id: String, text: String?) {
+    let directID = values.option("optionID")?.trimmingCharacters(in: .whitespacesAndNewlines)
+    let indexValue = values.optionInt64("optionIndex")
+    let textValue = values.option("option")?.trimmingCharacters(in: .whitespacesAndNewlines)
+    try validateOptionSelector(values)
     let options = try store.pollOptions(guid: pollGuid)
     guard !options.isEmpty else {
       throw ParsedValuesError.invalidOption("poll (could not decode options for \(pollGuid))")

Tests/imsgTests/BridgeCommandRegistrationTests.swift

@@ -238,6 +238,37 @@ func bridgeReplySendsKeepAssociatedMessageFallback() throws {
   }
 }
 
+@Test
+func bridgeV2InboxClaimsRequestBeforeDispatch() throws {
+  let testFile = URL(fileURLWithPath: #filePath)
+  let repoRoot =
+    testFile
+    .deletingLastPathComponent()
+    .deletingLastPathComponent()
+    .deletingLastPathComponent()
+  let helper = repoRoot.appendingPathComponent("Sources/IMsgHelper/IMsgInjected.m")
+  let source = stripObjectiveCComments(try String(contentsOf: helper, encoding: .utf8))
+  let processBody = try #require(functionBody(named: "processV2InboxFile", in: source))
+  let cleanupBody = try #require(functionBody(named: "cleanupOrphanedV2Claims", in: source))
+  let scanBody = try #require(functionBody(named: "scanV2Inbox", in: source))
+  let claim = try #require(
+    processBody.range(of: "rename(inPath.UTF8String, claimPath.UTF8String)"))
+  let read = try #require(
+    processBody.range(of: "dataWithContentsOfFile:claimPath"))
+  let dispatch = try #require(processBody.range(of: "processV2Envelope(envelope)"))
+
+  #expect(processBody.contains(#"@"%@.processing.%d""#))
+  #expect(processBody.contains("claimErrno != ENOENT"))
+  #expect(processBody.contains("removeItemAtPath:claimPath"))
+  #expect(!processBody.contains("processedRpcIds"))
+  #expect(claim.lowerBound < read.lowerBound)
+  #expect(read.lowerBound < dispatch.lowerBound)
+  #expect(cleanupBody.contains("kill(ownerPID, 0)"))
+  #expect(cleanupBody.contains("kV2ClaimMaxAge"))
+  #expect(cleanupBody.contains("removeItemAtPath:path"))
+  #expect(scanBody.contains("cleanupOrphanedV2Claims(entries)"))
+}
+
 @Test
 func injectedHelperWiresNativePollSend() throws {
   let testFile = URL(fileURLWithPath: #filePath)
@@ -361,6 +392,9 @@ func injectedHelperConstructorOnlySchedulesDelayedBootstrap() throws {
   let source = stripObjectiveCComments(try String(contentsOf: helper, encoding: .utf8))
   let constructorBody = try #require(functionBody(named: "injectedInit", in: source))
   let bootstrapBody = try #require(functionBody(named: "bridgeBootstrap", in: source))
+  let cleanupBody = try #require(functionBody(named: "injectedCleanup", in: source))
+  let bundleGuard = try #require(bootstrapBody.range(of: "com.apple.MobileSMS"))
+  let initializePaths = try #require(bootstrapBody.range(of: "initFilePaths()"))
 
   #expect(constructorBody.contains("dispatch_after"))
   #expect(constructorBody.contains("dispatch_async"))
@@ -374,10 +408,13 @@ func injectedHelperConstructorOnlySchedulesDelayedBootstrap() throws {
 
   #expect(bootstrapBody.contains("dispatch_once"))
   #expect(bootstrapBody.contains("@autoreleasepool"))
+  #expect(bundleGuard.lowerBound < initializePaths.lowerBound)
+  #expect(bootstrapBody.contains("bridgeDidBootstrap = YES"))
   #expect(bootstrapBody.contains("connectToDaemon"))
   #expect(bootstrapBody.contains("startFileWatcher()"))
   #expect(bootstrapBody.contains("startV2InboxWatcher()"))
   #expect(bootstrapBody.contains("registerEventObservers()"))
+  #expect(cleanupBody.contains("if (!bridgeDidBootstrap) return;"))
 }
 
 private func stripObjectiveCComments(_ source: String) -> String {