u

src/services/conversationManager.js

@@ -1,17 +1,18 @@
 import { LLMOrbiter, OpenRouterProvider } from 'libllmorbiter';
 import * as db from '../db/modelconversation.js';
 import * as dbCompanySetting from '../db/companysetting.js';
+import { getStore } from '../db/store.js';
 import env from '../config/env.js';
 import { emit as emitCompanyEvent } from '../api/sse.js';
-import { readFileSync } from 'fs';
-import { resolve, dirname } from 'path';
+import { readFileSync, writeFileSync, mkdirSync } from 'fs';
+import { resolve, dirname, extname } from 'path';
 import { fileURLToPath } from 'url';
 import { randomUUID } from 'crypto';
 import { createDashbackRequestTool } from '../tools/dashback-request.js';
 import { createAskUserTool } from '../tools/ask-user.js';
 import { createJavascriptExecuteTool } from '../tools/javascript.js';
 import { createResetVfsTool } from '../tools/reset-vfs.js';
-import { resolveVfsRoot } from '../tools/vfs.js';
+import { resolveVfsRoot, prepareVfs, getMimeType } from '../tools/vfs.js';
 import { listSkillFiles } from '../tools/skill.js';
 import { createApiSuggestionsOrIssuesTool } from '../tools/apisuggestionsorissues.js';
 import { createFindFilesTool } from '../tools/find-files.js';
@@ -129,6 +130,7 @@ async function buildSystemPrompt(vfsRoot, googleAuthId) {
         'VIRTUAL FILESYSTEM:',
         '- Persistent per-company VFS. Paths are root-relative forward-slash (e.g. "labels/sheet1.png").',
         '- Text: list with find_files, read with read_file, write with patch.',
+        '- User attachments are saved to VFS (e.g. "attachments/...") and tagged in user messages as `[ATTACHMENT name= path= type=]`. Inspect attachments with appropriate tools: `read_file` (text/csv/json), `pdf_read` (PDFs), or `analyze_image` (images).',
         '',
         'SKILL FILES:',
         '- Reusable instruction docs: "<slug>.SKILL" at VFS root. First line is title, followed by markdown.',
@@ -207,6 +209,107 @@ export function formatTimePrefix(createdDate) {
     return `[${hour}:${minute}${tz}] `;
 }
 
+export function parseIncomingUserMessage({ message, attachments: directAttachments, content: directContent } = {}) {
+    let text = '';
+    const attachments = Array.isArray(directAttachments) ? [...directAttachments] : [];
+
+    const tryParseJson = (str) => {
+        if (typeof str !== 'string') return null;
+        const trimmed = str.trim();
+        if ((trimmed.startsWith('{') && trimmed.endsWith('}')) || (trimmed.startsWith('[') && trimmed.endsWith(']'))) {
+            try {
+                return JSON.parse(trimmed);
+            } catch {
+                return null;
+            }
+        }
+        return null;
+    };
+
+    let candidate = message !== undefined ? message : directContent;
+
+    if (typeof candidate === 'string') {
+        const parsed = tryParseJson(candidate);
+        if (parsed && typeof parsed === 'object') {
+            candidate = parsed;
+        } else {
+            text = candidate;
+        }
+    }
+
+    if (candidate && typeof candidate === 'object' && !Array.isArray(candidate)) {
+        if (typeof candidate.content === 'string') {
+            const innerParsed = tryParseJson(candidate.content);
+            if (innerParsed && typeof innerParsed === 'object') {
+                if (typeof innerParsed.content === 'string') {
+                    text = innerParsed.content;
+                }
+                if (Array.isArray(innerParsed.attachments)) {
+                    attachments.push(...innerParsed.attachments);
+                }
+            } else {
+                text = candidate.content;
+            }
+        } else if (typeof candidate.message === 'string') {
+            const innerParsed = tryParseJson(candidate.message);
+            if (innerParsed && typeof innerParsed === 'object') {
+                if (typeof innerParsed.content === 'string') {
+                    text = innerParsed.content;
+                }
+                if (Array.isArray(innerParsed.attachments)) {
+                    attachments.push(...innerParsed.attachments);
+                }
+            } else {
+                text = candidate.message;
+            }
+        }
+
+        if (Array.isArray(candidate.attachments)) {
+            attachments.push(...candidate.attachments);
+        }
+    }
+
+    const uniqueAttachments = [];
+    const seen = new Set();
+    for (const att of attachments) {
+        if (!att || typeof att !== 'object') continue;
+        const key = att.id || `${att.name || att.filename}_${att.size || ''}`;
+        if (!seen.has(key)) {
+            seen.add(key);
+            uniqueAttachments.push(att);
+        }
+    }
+
+    return {
+        content: String(text || '').trim(),
+        attachments: uniqueAttachments,
+    };
+}
+
+export function cleanUserMessageContent(rawContent) {
+    if (!rawContent || typeof rawContent !== 'string') return '';
+    if (rawContent.includes('"attachments"') && (rawContent.trim().startsWith('{') || rawContent.includes('{"content"'))) {
+        try {
+            const parsed = parseIncomingUserMessage({ message: rawContent });
+            if (parsed.attachments.length > 0) {
+                const tags = parsed.attachments.map((att) => {
+                    const name = (att.name || att.filename || 'attachment').replace(/^.*[\\\/]/, '').replace(/[\0\r\n"]/g, '').trim() || 'attachment';
+                    const mime = att.mimeType || att.type || getMimeType(name);
+                    const vfsRelPath = `attachments/${name}`;
+                    const formatAttr = (val) => String(val).includes(' ') ? `"${val}"` : String(val);
+                    return `[ATTACHMENT name=${formatAttr(name)} path=${formatAttr(vfsRelPath)} type=${formatAttr(mime)}]`;
+                });
+                const prefix = tags.join(' ');
+                return parsed.content ? `${prefix}\n\n${parsed.content}` : prefix;
+            }
+            if (parsed.content) return parsed.content;
+        } catch {
+            // keep rawContent if parsing fails
+        }
+    }
+    return rawContent;
+}
+
 export function historyToLlmMessages(history) {
     const messages = [];
     for (const m of history) {
@@ -219,10 +322,11 @@ export function historyToLlmMessages(history) {
         }
         if (m.role === 'user') {
             const rawContent = payload.content ?? (typeof m.message === 'string' ? m.message : '');
+            const cleanedContent = cleanUserMessageContent(rawContent);
             const timePrefix = formatTimePrefix(m.created_at);
-            const content = /^\s*\[\d{2}:\d{2}[A-Za-z]+\]\s*/.test(rawContent)
-                ? rawContent
-                : `${timePrefix}${rawContent}`;
+            const content = /^\s*\[\d{2}:\d{2}[A-Za-z]+\]\s*/.test(cleanedContent)
+                ? cleanedContent
+                : `${timePrefix}${cleanedContent}`;
             messages.push({ role: 'user', content });
         } else if (m.role === 'assistant') {
             messages.push({
@@ -340,24 +444,143 @@ export class ConversationManager {
     }
 
     /**
-     * Handle an incoming user message.
-     * 1. Persists the user message (DB refuses unless `idle`).
-     * 2. Broadcasts it, marks the conversation `busy` (DB) and `thinking` (SSE).
-     * 3. Starts the LLM run in the background; the HTTP request returns immediately.
+     * Saves attachments sent with a user message to the company's VFS
+     * under `attachments/<filename>` and registers them in `modelvfs`.
      *
      * @param {number} googleAuthId
-     * @param {{ message: string }} input
+     * @param {Array<object>} attachments
+     * @returns {Promise<Array<string>>} Array of `[ATTACHMENT name= path= type=]` tags
+     */
+    async saveAttachmentsToVfs(googleAuthId, attachments) {
+        if (!Array.isArray(attachments) || attachments.length === 0) return [];
+
+        let vfsRoot;
+        try {
+            vfsRoot = await prepareVfs({ googleAuthId });
+        } catch (err) {
+            console.error(`[ConversationManager ${this.conversationId}] prepareVfs error:`, err);
+            return [];
+        }
+
+        const store = await getStore();
+        const attachmentsDir = resolve(vfsRoot, 'attachments');
+        try {
+            mkdirSync(attachmentsDir, { recursive: true });
+        } catch (err) {
+            console.error(`[ConversationManager ${this.conversationId}] failed to create attachments directory:`, err);
+        }
+
+        const tags = [];
+        const usedNames = new Set();
+
+        for (const att of attachments) {
+            if (!att || typeof att !== 'object') continue;
+
+            const rawName = (att.name || att.filename || `attachment-${Date.now()}`).replace(/^.*[\\\/]/, '');
+            const cleanName = rawName.replace(/[\0\r\n"]/g, '').trim() || 'attachment';
+
+            let finalName = cleanName;
+            let counter = 1;
+            const ext = extname(cleanName);
+            const base = ext ? cleanName.slice(0, -ext.length) : cleanName;
+            while (usedNames.has(finalName.toLowerCase())) {
+                counter++;
+                finalName = `${base}_${counter}${ext}`;
+            }
+            usedNames.add(finalName.toLowerCase());
+
+            let fileBuffer = null;
+            if (att.dataUrl && typeof att.dataUrl === 'string') {
+                const commaIdx = att.dataUrl.indexOf(',');
+                if (commaIdx !== -1) {
+                    const metaPart = att.dataUrl.slice(0, commaIdx);
+                    const dataPart = att.dataUrl.slice(commaIdx + 1);
+                    if (metaPart.includes(';base64')) {
+                        fileBuffer = Buffer.from(dataPart, 'base64');
+                    } else {
+                        fileBuffer = Buffer.from(decodeURIComponent(dataPart), 'utf8');
+                    }
+                }
+            }
+            if (!fileBuffer && att.textContent != null) {
+                fileBuffer = Buffer.from(String(att.textContent), 'utf8');
+            }
+            if (!fileBuffer && att.content != null) {
+                fileBuffer = Buffer.isBuffer(att.content) ? att.content : Buffer.from(String(att.content));
+            }
+
+            let mime = att.mimeType || att.type || '';
+            if (!mime && att.dataUrl && typeof att.dataUrl === 'string') {
+                const match = att.dataUrl.match(/^data:([^;,]+)/);
+                if (match) mime = match[1];
+            }
+            if (!mime) {
+                mime = getMimeType(finalName);
+            }
+
+            const vfsRelPath = `attachments/${finalName}`;
+            const diskPath = resolve(vfsRoot, vfsRelPath);
+
+            if (fileBuffer) {
+                try {
+                    writeFileSync(diskPath, fileBuffer);
+                } catch (err) {
+                    console.error(`[ConversationManager ${this.conversationId}] failed to write attachment disk file:`, err);
+                }
+
+                try {
+                    await store.modelvfs.write(googleAuthId, {
+                        name: vfsRelPath,
+                        mime,
+                        content: fileBuffer,
+                    });
+                } catch (err) {
+                    console.error(`[ConversationManager ${this.conversationId}] failed to write attachment to modelvfs:`, err);
+                }
+            }
+
+            const formatAttr = (val) => String(val).includes(' ') ? `"${val}"` : String(val);
+            tags.push(`[ATTACHMENT name=${formatAttr(finalName)} path=${formatAttr(vfsRelPath)} type=${formatAttr(mime)}]`);
+        }
+
+        return tags;
+    }
+
+    /**
+     * Handle an incoming user message.
+     * 1. Extracts text content and saves any attachments into the company VFS (`attachments/`).
+     * 2. Prefixes user message with `[ATTACHMENT name= path= type=]` tags.
+     * 3. Persists the user message (DB refuses unless `idle`).
+     * 4. Broadcasts it, marks the conversation `busy` (DB) and `thinking` (SSE).
+     * 5. Starts the LLM run in the background; the HTTP request returns immediately.
+     *
+     * @param {number} googleAuthId
+     * @param {{ message?: any, attachments?: Array<object>, content?: any }} input
      * @returns {Promise<object>} The created user message
      * @throws {Error} 'Conversation is not idle' while a run is active.
      */
-    async handleUserMessage(googleAuthId, { message }) {
+    async handleUserMessage(googleAuthId, { message, attachments, content }) {
         if (this.isProcessing) throw new Error('Conversation is not idle');
         this.isProcessing = true;
 
         let userMsg;
         try {
-            const rawContent = typeof message === 'string' ? message : (message?.content || '');
-            const payload = JSON.stringify({ content: rawContent });
+            const parsed = parseIncomingUserMessage({ message, attachments, content });
+            if (!parsed.content && parsed.attachments.length === 0) {
+                throw new Error('Message required');
+            }
+
+            let finalMessage = parsed.content;
+
+            if (parsed.attachments.length > 0) {
+                const tags = await this.saveAttachmentsToVfs(googleAuthId, parsed.attachments);
+                if (tags.length > 0) {
+                    const prefix = tags.join(' ');
+                    finalMessage = parsed.content ? `${prefix}\n\n${parsed.content}` : prefix;
+                }
+            }
+
+            const payload = JSON.stringify({ content: finalMessage });
             userMsg = await db.addMessage(googleAuthId, this.conversationId, { message: payload, role: 'user' });
             await db.setState(googleAuthId, this.conversationId, 'busy');
         } catch (err) {