diff --git a/LifeOS/install/LIFEOS/PULSE/lib/imessage-send.ts b/LifeOS/install/LIFEOS/PULSE/lib/imessage-send.ts index 46a905e697..0e419c6f68 100644 --- a/LifeOS/install/LIFEOS/PULSE/lib/imessage-send.ts +++ b/LifeOS/install/LIFEOS/PULSE/lib/imessage-send.ts @@ -25,13 +25,15 @@ export async function sendMessage( const chunks = splitMessage(text, MAX_MESSAGE_LENGTH) for (const chunk of chunks) { - const escaped = escapeForAppleScript(chunk) + // Zero-width-space prefix marks this as a bot message so the poller + // can ignore its own echo in message-to-self conversations. + const escaped = escapeForAppleScript("\u200B" + chunk) // Use buddy-based send — works reliably on modern macOS const script = ` tell application "Messages" set targetService to 1st account whose service type = iMessage - set targetBuddy to participant targetService handle "${escapeForAppleScript(handle)}" + set targetBuddy to participant "${escapeForAppleScript(handle)}" of targetService send "${escaped}" to targetBuddy end tell` diff --git a/LifeOS/install/LIFEOS/PULSE/lib/messages-db.ts b/LifeOS/install/LIFEOS/PULSE/lib/messages-db.ts index 27f5261319..65f3588361 100644 --- a/LifeOS/install/LIFEOS/PULSE/lib/messages-db.ts +++ b/LifeOS/install/LIFEOS/PULSE/lib/messages-db.ts @@ -19,6 +19,12 @@ const CHAT_DB_PATH = join(HOME, "Library", "Messages", "chat.db") // chat.db stores dates as nanoseconds since Apple epoch const APPLE_EPOCH_OFFSET = 978307200 // seconds between Unix epoch and Apple epoch +export interface MessageAttachment { + filename: string // absolute path under ~/Library/Messages/Attachments + mimeType: string + transferName: string // original filename as sent +} + export interface IncomingMessage { rowid: number text: string @@ -26,11 +32,19 @@ export interface IncomingMessage { handle: string // phone number or email service: string // "iMessage" or "SMS" chatId: string // chat identifier for reply routing + isFromMe: boolean + attachments: MessageAttachment[] } /** - * Get new incoming messages since a given ROWID. - * Only returns messages NOT from the user (is_from_me = 0). + * Get new messages since a given ROWID. + * + * Returns BOTH directions: is_from_me=0 (received) and is_from_me=1 (sent). + * Sent rows are needed for the message-to-self setup: the principal texts + * their own handle, and the iCloud "received echo" copy (is_from_me=0) is + * NOT reliably generated — sometimes it arrives, sometimes never. The + * module filters sent rows down to the allowlisted self-chat and dedups + * against echo copies. */ export function getNewMessages(sinceRowId: number): IncomingMessage[] { const db = new Database(CHAT_DB_PATH, { readonly: true }) @@ -51,14 +65,15 @@ export function getNewMessages(sinceRowId: number): IncomingMessage[] { LEFT JOIN chat_message_join cmj ON cmj.message_id = m.ROWID LEFT JOIN chat c ON cmj.chat_id = c.ROWID WHERE m.ROWID > ? - AND m.is_from_me = 0 - AND m.text IS NOT NULL - AND m.text != '' + AND ( + (m.text IS NOT NULL AND m.text != '') + OR EXISTS (SELECT 1 FROM message_attachment_join maj WHERE maj.message_id = m.ROWID) + ) ORDER BY m.ROWID ASC `) .all(sinceRowId) as Array<{ rowid: number - text: string + text: string | null date: number is_from_me: number handle: string @@ -66,14 +81,37 @@ export function getNewMessages(sinceRowId: number): IncomingMessage[] { chat_id: string }> + const attachmentQuery = db.query(` + SELECT + COALESCE(a.filename, '') as filename, + COALESCE(a.mime_type, '') as mime_type, + COALESCE(a.transfer_name, '') as transfer_name + FROM message_attachment_join maj + JOIN attachment a ON a.ROWID = maj.attachment_id + WHERE maj.message_id = ? + `) + return rows.map((row) => ({ rowid: row.rowid, - text: row.text, + // Strip U+FFFC object-replacement placeholders that stand in for attachments + text: (row.text ?? "").replace(//g, "").trim(), handle: row.handle, service: row.service, chatId: row.chat_id, + isFromMe: row.is_from_me === 1, // Convert Apple nanosecond timestamp to JS Date date: new Date((row.date / 1e9 + APPLE_EPOCH_OFFSET) * 1000), + attachments: (attachmentQuery.all(row.rowid) as Array<{ + filename: string + mime_type: string + transfer_name: string + }>) + .filter((a) => a.filename) + .map((a) => ({ + filename: a.filename.replace(/^~\//, HOME + "/"), + mimeType: a.mime_type, + transferName: a.transfer_name, + })), })) } finally { db.close() diff --git a/LifeOS/install/LIFEOS/PULSE/modules/imessage.ts b/LifeOS/install/LIFEOS/PULSE/modules/imessage.ts index 0a6f3f78f3..cacf381523 100644 --- a/LifeOS/install/LIFEOS/PULSE/modules/imessage.ts +++ b/LifeOS/install/LIFEOS/PULSE/modules/imessage.ts @@ -23,7 +23,9 @@ import { getNewMessages, getLatestRowId, verifyAccess, + type IncomingMessage, } from "../lib/messages-db" +import { $ } from "bun" import { sendMessage } from "../lib/imessage-send" import { join } from "path" import { appendFile, mkdir, rename } from "fs/promises" @@ -78,6 +80,8 @@ let processing = false let lastError: string | undefined let allowedHandles = new Set() let pollIntervalMs = 3000 +// Insertion-ordered set for echo dedup (sent row vs delayed iCloud echo copy) +const recentlyProcessed = new Set() let maxTurns = 25 let sdkTimeoutMs = 120_000 let conversationStore: ConversationStore | null = null @@ -127,14 +131,64 @@ async function appendChatLog( await appendFile(chatLogPath, entry).catch(() => {}) } +// ── Attachment Staging ── +// +// Attachment files live under ~/Library/Messages/Attachments, which is +// TCC-protected: this bun process has Full Disk Access, but the claude +// subprocess it spawns does not. Copy each attachment into our own state +// dir so the SDK session's Read tool can view it, converting HEIC→JPEG +// (Read handles jpeg/png natively, not HEIC). + +const MAX_ATTACHMENTS = 6 + +async function stageAttachments( + msg: IncomingMessage, +): Promise { + const staged: string[] = [] + const dir = join(STATE_DIR, "attachments", String(msg.rowid)) + for (const att of msg.attachments.slice(0, MAX_ATTACHMENTS)) { + try { + const src = Bun.file(att.filename) + if (!(await src.exists())) continue + await mkdir(dir, { recursive: true }) + const base = (att.transferName || att.filename.split("/").pop() || "file") + .replace(/[^\w.-]/g, "_") + const isHeic = /heic|heif/i.test(att.mimeType) || /\.heic$|\.heif$/i.test(att.filename) + if (isHeic) { + const dst = join(dir, base.replace(/\.[^.]*$/, "") + ".jpg") + await $`sips -s format jpeg ${att.filename} --out ${dst}`.quiet() + staged.push(dst) + } else { + const dst = join(dir, base) + await Bun.write(dst, src) + staged.push(dst) + } + } catch (err) { + log("warn", "Failed to stage attachment", { + file: att.filename, + error: String(err), + }) + } + } + if (msg.attachments.length > MAX_ATTACHMENTS) { + log("warn", "Attachment count capped", { + total: msg.attachments.length, + staged: MAX_ATTACHMENTS, + }) + } + return staged +} + // ── Process a Single Message ── async function processMessage( text: string, handle: string, + stagedAttachments: string[] = [], ): Promise { - const sanitized = sanitize(text) - if (!sanitized) return "" + let sanitized = sanitize(text) + if (!sanitized && stagedAttachments.length === 0) return "" + if (!sanitized) sanitized = "(no text — attachment only)" const injection = analyzeForInjection(sanitized) if (injection.riskLevel === "CRITICAL") { @@ -156,6 +210,10 @@ async function processMessage( prompt = `Previous conversation:\n${historyText}\n\nPrincipal's new message: ${sanitized}` } + if (stagedAttachments.length > 0) { + prompt += `\n\n[The principal attached ${stagedAttachments.length} file(s) to this message. View them with the Read tool:\n${stagedAttachments.map((p) => `- ${p}`).join("\n")}]` + } + // settingSources does NOT cover MCP config (public issue #1553, // @MatiasBarboza) — without an explicit mcpServers option this session gets // zero MCP servers. Default-deny by design on a remote channel; opt in per @@ -277,6 +335,11 @@ ${mcpStatusPromptLine(Object.keys(remoteMcp))}`, // ── Poll Loop ── async function poll() { + // A previous poll's message is still in its SDK session. Don't read — + // leave the cursor where it is so pending messages stay in the DB and + // get picked up (in order) on the next tick instead of being dropped. + if (processing) return + try { const messages = getNewMessages(lastRowId) @@ -284,36 +347,84 @@ async function poll() { // Update cursor regardless of auth lastRowId = msg.rowid - // Auth check - if (!allowedHandles.has(msg.handle)) { + // Self-chat echo guard: our own replies carry a zero-width-space + // marker and come back as is_from_me=0 copies in a message-to-self + // conversation — never answer them or the bot loops on itself. + if (msg.text.startsWith("\u200B")) { + continue + } + + // Auth check. Received rows must come from an allowlisted handle. + // Sent rows (is_from_me=1) qualify ONLY inside the allowlisted + // self-chat — the principal texting their own handle — because the + // iCloud received-echo copy is not reliably generated. Sent messages + // to anyone else are the principal's private traffic: skip silently. + if (msg.isFromMe) { + if (!allowedHandles.has(msg.chatId) && !allowedHandles.has(msg.handle)) { + continue + } + } else if (!allowedHandles.has(msg.handle)) { log("warn", "Rejected message from unauthorized handle", { handle: msg.handle, }) continue } + // Nothing to respond to (e.g. reaction/sticker rows with no content) + if (!msg.text && msg.attachments.length === 0) { + continue + } + + // Freshness gate: iCloud sync can re-INSERT old messages with brand + // new ROWIDs (observed 2026-08-13: yesterday's full history re-synced + // overnight and the cursor treated it all as new — the bot answered + // ~240 stale messages). ROWID order is not arrival order of new + // content; the message's own timestamp is the truth. Never answer + // anything older than this window. + const ageMs = Date.now() - msg.date.getTime() + if (ageMs > 15 * 60 * 1000) { + log("info", "Skipping stale message (iCloud re-sync)", { + rowid: msg.rowid, + ageMinutes: Math.round(ageMs / 60000), + }) + continue + } + + // Echo dedup: when iCloud DOES deliver the received copy of a + // self-chat message, the same content appears twice (sent row + + // echo row, identical timestamp). Process it once. + const dedupKey = `${msg.date.getTime()}|${msg.text}|${msg.attachments.length}` + if (recentlyProcessed.has(dedupKey)) { + continue + } + recentlyProcessed.add(dedupKey) + if (recentlyProcessed.size > 200) { + const oldest = recentlyProcessed.values().next().value + if (oldest !== undefined) recentlyProcessed.delete(oldest) + } + messagesReceived++ log("info", "Message received", { handle: msg.handle, textLength: msg.text.length, + attachments: msg.attachments.length, rowid: msg.rowid, }) - // Sequential processing - if (processing) { - await sendMessage( - msg.handle, - "Still processing your previous message. Please wait.", - ) - continue - } + // Sent rows in the self-chat can carry an empty handle; route the + // reply to the chat identifier (the principal's own handle) then. + const replyTo = allowedHandles.has(msg.handle) ? msg.handle : msg.chatId processing = true const startTime = Date.now() try { - const response = await processMessage(msg.text, msg.handle) - const sent = await sendMessage(msg.handle, response) + const staged = await stageAttachments(msg) + if (staged.length > 0) { + log("info", "Attachments staged", { count: staged.length, rowid: msg.rowid }) + } + const response = await processMessage(msg.text, replyTo, staged) + const sent = await sendMessage(replyTo, response) if (sent) { messagesResponded++ @@ -333,7 +444,7 @@ async function poll() { lastError = String(err) log("error", "Message processing failed", { error: lastError }) await sendMessage( - msg.handle, + replyTo, "Something went wrong processing your message. Try again?", ).catch(() => {}) } finally {