agent responses to messages

AGENTS.md

@@ -1,9 +1,10 @@
 # Agent Instructions
 
-Desktop Agent Harness. Swift 6.4+, Xcode 27, macOS 27. You must follow these instructions specifically.
+You are an Apple Swift and SwiftUI expoert building a Desktop Agent Harness using Swift 6.4+, Xcode 27, macOS 27. You must follow these instructions specifically.
 
 ## Before changing code
 
+- You are kind, slow and methodical. You do not rush.
 - Read the files on the code path you are changing. Do not guess.
 - Check `Info.plist` and app configuration before assuming a code bug.
 - When fixing issues do not assume. Make an assertion about where the problem is, confirm your assertion is true, and then fix the issue there.

packages/DBRepository/Sources/DBRepository/DBRepositoryMessaging.swift

@@ -231,6 +231,52 @@ public extension DBRepository {
         }
     }
 
+    /// Inbound messages that have not yet been claimed for an agent turn.
+    func listUnclaimedInboundMessagingMessages(limit: Int = 40) throws -> [MessagingPersistResult] {
+        let pageSize = max(1, min(limit, 80))
+        return try withDatabaseHandle { handle in
+            let sql = """
+            SELECT
+                m.id, m.thread_id, m.vendor_message_id, m.direction, m.sender, m.body, m.created_at,
+                m.parent_vendor_message_id, m.reply_count,
+                t.id, t.plugin_id, t.vendor_thread_id, t.title, t.last_activity_at, t.muted,
+                t.unread_count, t.created_at
+            FROM messaging_messages m
+            INNER JOIN messaging_threads t ON t.id = m.thread_id
+            LEFT JOIN messaging_agent_handled h
+              ON h.plugin_id = t.plugin_id
+             AND h.vendor_message_id = m.vendor_message_id
+            WHERE m.direction = \(quoted(MessagingMessageDirection.inbound.rawValue))
+              AND m.vendor_message_id IS NOT NULL
+              AND TRIM(m.vendor_message_id) != ''
+              AND h.plugin_id IS NULL
+            ORDER BY m.created_at DESC, m.id DESC
+            LIMIT \(pageSize);
+            """
+            var statement: OpaquePointer?
+            guard sqlite3_prepare_v2(handle, sql, -1, &statement, nil) == SQLITE_OK, let statement else {
+                throw Self.sqliteError(handle: handle, fallback: "Failed to prepare unclaimed inbound messaging list.")
+            }
+            defer { sqlite3_finalize(statement) }
+            var rows: [MessagingPersistResult] = []
+            while sqlite3_step(statement) == SQLITE_ROW {
+                let message = try decodeMessagingMessage(statement: statement)
+                let thread = MessagingThreadDTO(
+                    id: try columnString(statement, index: 9),
+                    pluginID: try columnString(statement, index: 10),
+                    vendorThreadID: try columnString(statement, index: 11),
+                    title: try columnString(statement, index: 12),
+                    lastActivityAt: Self.iso8601Formatter().date(from: try columnString(statement, index: 13)) ?? .now,
+                    muted: sqlite3_column_int(statement, 14) != 0,
+                    unreadCount: Int(sqlite3_column_int(statement, 15)),
+                    createdAt: Self.iso8601Formatter().date(from: try columnString(statement, index: 16)) ?? .now
+                )
+                rows.append(MessagingPersistResult(inserted: true, message: message, thread: thread))
+            }
+            return rows
+        }
+    }
+
     /// Latest reply body for each parent, for the channel "N replies" row.
     func latestReplyPreviews(
         threadID: String,

packages/DBRepository/Sources/DBRepository/DBRepositoryMessagingAgentHandled.swift

@@ -39,4 +39,45 @@ public extension DBRepository {
             }
         }
     }
+
+    func releaseMessagingAgentHandling(pluginID: String, vendorMessageID: String) throws {
+        let trimmedPluginID = pluginID.trimmingCharacters(in: .whitespacesAndNewlines)
+        let trimmedMessageID = vendorMessageID.trimmingCharacters(in: .whitespacesAndNewlines)
+        guard !trimmedPluginID.isEmpty, !trimmedMessageID.isEmpty else { return }
+        try withDatabaseHandle { handle in
+            try Self.execute("""
+            DELETE FROM messaging_agent_handled
+            WHERE plugin_id = \(quoted(trimmedPluginID))
+              AND vendor_message_id = \(quoted(trimmedMessageID));
+            """, on: handle)
+        }
+    }
+
+    /// Lets `$profile` inbound retry when a turn was claimed but never posted a reply.
+    func releaseUnansweredProfileTokenClaims() throws {
+        try withDatabaseHandle { handle in
+            try Self.execute("""
+            DELETE FROM messaging_agent_handled
+            WHERE rowid IN (
+                SELECT h.rowid
+                FROM messaging_agent_handled h
+                INNER JOIN messaging_threads t ON t.plugin_id = h.plugin_id
+                INNER JOIN messaging_messages m
+                  ON m.thread_id = t.id
+                 AND m.vendor_message_id = h.vendor_message_id
+                WHERE m.direction = \(quoted(MessagingMessageDirection.inbound.rawValue))
+                  AND (
+                    TRIM(m.body) LIKE '$%'
+                    OR m.body LIKE '%$%'
+                  )
+                  AND NOT EXISTS (
+                    SELECT 1 FROM messaging_messages o
+                    WHERE o.thread_id = m.thread_id
+                      AND o.direction = \(quoted(MessagingMessageDirection.outbound.rawValue))
+                      AND o.created_at >= m.created_at
+                  )
+            );
+            """, on: handle)
+        }
+    }
 }

packages/DBRepository/Tests/DBRepositoryTests/DBMessagingAgentHandledTests.swift

@@ -19,6 +19,93 @@ final class DBMessagingAgentHandledTests: XCTestCase {
         XCTAssertFalse(secondClaim)
     }
 
+    func testListUnclaimedInboundExcludesClaimedRows() async throws {
+        let repository = try makeRepository()
+        _ = try await repository.createEmptyDatabaseIfNeeded(username: "app-user", password: "app-secret")
+        try await repository.upsertMessagingConnector(
+            MessagingConnectorDTO(pluginID: "slack-connection", displayName: "Slack")
+        )
+        _ = try await repository.persistMessagingInbound(
+            MessagingInboundRecord(
+                pluginID: "slack-connection",
+                vendorThreadID: "C123",
+                threadTitle: "#general",
+                vendorMessageID: "1710000002.000200",
+                sender: "U07FKG8DV19",
+                body: "$orchestrator tell me about yourself"
+            )
+        )
+        _ = try await repository.persistMessagingInbound(
+            MessagingInboundRecord(
+                pluginID: "slack-connection",
+                vendorThreadID: "C123",
+                threadTitle: "#general",
+                vendorMessageID: "1710000002.000201",
+                sender: "U07FKG8DV19",
+                body: "hello without mention"
+            )
+        )
+
+        let unclaimed = try await repository.listUnclaimedInboundMessagingMessages()
+        XCTAssertEqual(Set(unclaimed.compactMap(\.message.vendorMessageID)), [
+            "1710000002.000200",
+            "1710000002.000201"
+        ])
+
+        let claimed = try await repository.claimMessagingAgentHandling(
+            pluginID: "slack-connection",
+            vendorMessageID: "1710000002.000200"
+        )
+        XCTAssertTrue(claimed)
+        let remaining = try await repository.listUnclaimedInboundMessagingMessages()
+        XCTAssertEqual(remaining.compactMap(\.message.vendorMessageID), ["1710000002.000201"])
+    }
+
+    func testReleaseUnansweredProfileTokenClaims() async throws {
+        let repository = try makeRepository()
+        _ = try await repository.createEmptyDatabaseIfNeeded(username: "app-user", password: "app-secret")
+        try await repository.upsertMessagingConnector(
+            MessagingConnectorDTO(pluginID: "slack-connection", displayName: "Slack")
+        )
+        _ = try await repository.persistMessagingInbound(
+            MessagingInboundRecord(
+                pluginID: "slack-connection",
+                vendorThreadID: "C123",
+                threadTitle: "#general",
+                vendorMessageID: "1710000002.000200",
+                sender: "U07FKG8DV19",
+                body: "$developer tell me about yourself"
+            )
+        )
+        let claimed = try await repository.claimMessagingAgentHandling(
+            pluginID: "slack-connection",
+            vendorMessageID: "1710000002.000200"
+        )
+        XCTAssertTrue(claimed)
+        try await repository.releaseUnansweredProfileTokenClaims()
+        let unclaimed = try await repository.listUnclaimedInboundMessagingMessages()
+        XCTAssertEqual(unclaimed.compactMap(\.message.vendorMessageID), ["1710000002.000200"])
+    }
+
+    func testReleaseMessagingAgentHandlingAllowsRetry() async throws {
+        let repository = try makeRepository()
+        _ = try await repository.createEmptyDatabaseIfNeeded(username: "app-user", password: "app-secret")
+        let first = try await repository.claimMessagingAgentHandling(
+            pluginID: "slack-connection",
+            vendorMessageID: "171.9"
+        )
+        XCTAssertTrue(first)
+        try await repository.releaseMessagingAgentHandling(
+            pluginID: "slack-connection",
+            vendorMessageID: "171.9"
+        )
+        let retry = try await repository.claimMessagingAgentHandling(
+            pluginID: "slack-connection",
+            vendorMessageID: "171.9"
+        )
+        XCTAssertTrue(retry)
+    }
+
     private func makeRepository() throws -> DBRepository {
         let directory = FileManager.default.temporaryDirectory
             .appendingPathComponent(UUID().uuidString, isDirectory: true)

packages/DerrickBackend/Sources/DerrickBackend/ConnectorMessagingCommandService.swift

@@ -102,6 +102,8 @@ public actor ConnectorMessagingCommandService {
                     parentVendorMessageID: request.parentVendorMessageID,
                     repository: repository
                 )
+                // Route on a detached task. Awaiting processInbound here deadlocks:
+                // the turn client sends through this same actor.
             case .send:
                 guard let vendorThreadID = request.vendorThreadID?.trimmingCharacters(in: .whitespacesAndNewlines),
                       !vendorThreadID.isEmpty,
@@ -129,6 +131,9 @@ public actor ConnectorMessagingCommandService {
                 request: request
             )
             DerrickMessagingInboundSignal.postRefresh()
+            if request.kind == .pollInbox {
+                DerrickMessagingIngressSignal.postPoll()
+            }
         } catch {
             finish(request.operationID, status: .failed, error: error.localizedDescription)
             await log(

packages/DerrickBackend/Sources/DerrickBackend/MessagingAgentIngressRouter.swift

@@ -1,6 +1,5 @@
 import DBRepository
 import Foundation
-import Plugin
 import Structure
 
 /// Routes newly persisted inbound connector messages to agent profiles when the bot is mentioned.
@@ -10,11 +9,29 @@ public enum MessagingAgentIngressRouter: Sendable {
         repository: DBRepository
     ) async {
         guard let routeHandler = InProcessServiceBridges.messagingAgentRoute else {
+            fputs("[MessagingAgentIngressRouter] no route handler installed — inbound agent turns are skipped\n", stderr)
             return
         }
 
-        for row in rows where row.inserted && row.message.direction == .inbound {
+        for row in rows where row.message.direction == .inbound {
             guard let route = await routeCandidate(from: row, repository: repository) else {
+                let body = row.message.body.trimmingCharacters(in: .whitespacesAndNewlines)
+                if AgentProfileTokenParser.parse(message: body).handle != nil {
+                    fputs(
+                        "[MessagingAgentIngressRouter] skipped \(body.prefix(80)) pluginID=\(row.thread.pluginID)\n",
+                        stderr
+                    )
+                }
+                let parsedHandle = AgentProfileTokenParser.parse(message: body).handle
+                if parsedHandle == nil,
+                   let vendorMessageID = row.message.vendorMessageID?
+                    .trimmingCharacters(in: .whitespacesAndNewlines),
+                   !vendorMessageID.isEmpty {
+                    _ = try? await repository.claimMessagingAgentHandling(
+                        pluginID: row.thread.pluginID,
+                        vendorMessageID: vendorMessageID
+                    )
+                }
                 continue
             }
             do {
@@ -23,12 +40,20 @@ public enum MessagingAgentIngressRouter: Sendable {
                     vendorMessageID: route.inboundVendorMessageID
                 )
                 guard claimed else { continue }
+                fputs(
+                    "[MessagingAgentIngressRouter] routing pluginID=\(route.pluginID) profile=\(route.profileHandle) message=\(route.inboundVendorMessageID)\n",
+                    stderr
+                )
                 try await routeHandler(route)
             } catch {
                 fputs(
                     "[MessagingAgentIngressRouter] route failed pluginID=\(route.pluginID) message=\(route.inboundVendorMessageID): \(error.localizedDescription)\n",
                     stderr
                 )
+                try? await repository.releaseMessagingAgentHandling(
+                    pluginID: route.pluginID,
+                    vendorMessageID: route.inboundVendorMessageID
+                )
                 await ServiceLogRecorder.shared.record(
                     service: "messaging",
                     level: .error,
@@ -49,61 +74,59 @@ public enum MessagingAgentIngressRouter: Sendable {
         guard !body.isEmpty else { return nil }
         guard !ConnectorMentionParser.isAutomatedOutboundEcho(body: body) else { return nil }
 
-        let manifestJSON = (try? await repository.listLatestPluginFactoryManifests()
-            .first(where: { $0.pluginID == row.thread.pluginID })?
-            .manifestJSON) ?? ""
-        guard supportsBotMentionRouting(manifestJSON: manifestJSON) else { return nil }
-
-        guard let botUserID = await SlackBotIdentityResolver.Cache.shared
+        let botUserID = await SlackBotIdentityResolver.Cache.shared
             .botUserID(pluginID: row.thread.pluginID)
-        else {
+            ?? ""
+        if !botUserID.isEmpty,
+           message.sender.trimmingCharacters(in: .whitespacesAndNewlines) == botUserID {
             return nil
         }
 
-        if message.sender.trimmingCharacters(in: .whitespacesAndNewlines) == botUserID {
+        guard let vendorMessageID = message.vendorMessageID?
+            .trimmingCharacters(in: .whitespacesAndNewlines),
+              !vendorMessageID.isEmpty
+        else {
             return nil
         }
 
+        let threadParent = ConnectorMentionParser.agentReplyThreadParentVendorMessageID(
+            inboundVendorMessageID: vendorMessageID,
+            existingParentVendorMessageID: message.parentVendorMessageID
+        )
+        var continuation: String?
+        if message.isReply {
+            let threadMessages = (try? await repository.listMessagingMessages(
+                threadID: row.thread.id,
+                limit: MessagingViewport.maxVisibleMessages,
+                filter: .replyThread(parentVendorMessageID: threadParent)
+            )) ?? []
+            continuation = ConnectorMentionParser.continuationProfileHandle(
+                in: threadMessages,
+                excludingVendorMessageID: vendorMessageID
+            )
+        }
+
         guard let resolved = ConnectorMentionParser.resolvePrompt(
             body: body,
             botUserID: botUserID,
+            continuationProfileHandle: continuation,
             channelDefaultProfileHandle: row.thread.defaultAgentProfileHandle,
             profileCatalog: (try? await profileCatalog(repository: repository)) ?? []
         ) else {
             return nil
         }
 
-        guard let vendorMessageID = message.vendorMessageID?
-            .trimmingCharacters(in: .whitespacesAndNewlines),
-              !vendorMessageID.isEmpty
-        else {
-            return nil
-        }
-
         return MessagingAgentRoute(
             pluginID: row.thread.pluginID,
             threadID: row.thread.id,
             vendorThreadID: row.thread.vendorThreadID,
-            parentVendorMessageID: message.parentVendorMessageID,
+            parentVendorMessageID: threadParent,
             inboundVendorMessageID: vendorMessageID,
             profileHandle: resolved.profileHandle,
             prompt: resolved.prompt
         )
     }
 
-    private static func supportsBotMentionRouting(manifestJSON: String) -> Bool {
-        guard let data = manifestJSON.data(using: .utf8),
-              let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
-              let extensions = object["extensions"] as? [String: Any],
-              let derrick = extensions["app.derrick"] as? [String: Any]
-        else {
-            return false
-        }
-        let role = (derrick["role"] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
-        let authScheme = (derrick["auth_scheme"] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
-        return role == "connector" && authScheme == "bot_token"
-    }
-
     private static func profileCatalog(repository: DBRepository) async throws -> [AgentProfileCatalogEntry] {
         try await repository.listAgentProfiles()
             .filter(\.isEnabled)