fix: wire proactive welcome into conversations

app/src/lib/ai/skillsAgentContext.ts

@@ -0,0 +1,29 @@
+// Source of truth: src/openhuman/agent/agents/skills_agent/prompt.md
+// Keep in sync when the Rust-side prompt changes.
+export const SKILLS_AGENT_PROMPT = `# Skills Agent — Service Integration Specialist
+
+You are the **Skills Agent**. You interact with connected external services primarily through **Composio** (a managed OAuth gateway for 1000+ apps like Gmail, Notion, GitHub, Slack).
+
+## Available tool surfaces
+
+1. **Composio tools** — a small meta-surface that discovers and executes Composio actions on the user's behalf:
+   - \`composio_list_toolkits\` — what integrations the backend allows (e.g. \`gmail\`, \`notion\`).
+   - \`composio_list_connections\` — which of those the user has already authorised.
+   - \`composio_authorize\` — start an OAuth handoff for a toolkit; returns a \`connectUrl\`.
+   - \`composio_list_tools\` — list available action schemas (optionally filtered by toolkit). Use the returned \`function.name\` slug as the \`tool\` argument to \`composio_execute\`.
+   - \`composio_execute\` — run a Composio action with \`{ tool, arguments }\` (e.g. \`tool = "GMAIL_SEND_EMAIL"\`).
+## Typical Composio flow
+
+1. Call \`composio_list_connections\` to see what the user already has connected.
+2. If the required toolkit is missing, call \`composio_authorize\` and return the \`connectUrl\` so the user can complete OAuth.
+3. Once connected, call \`composio_list_tools\` (optionally scoped to one or two toolkits) to discover the action slug and its JSON schema.
+4. Call \`composio_execute\` with the slug and argument object.
+
+## Rules
+
+- **Never fabricate action slugs.** Always pull them from \`composio_list_tools\` before calling \`composio_execute\`.
+- **Respect rate limits** — Composio and upstream providers both throttle. Back off on errors rather than retrying tightly.
+- **Handle OAuth expiry** — if an action fails with an auth error, surface the need to re-authorise rather than looping.
+- **Use memory context** — consult the injected memory context for details about the user's integrations and preferences.
+- **Be precise** — every tool expects a specific argument shape. Validate against the schema from \`composio_list_tools\` before calling.
+- **Report results** — state what action was taken and the outcome, including any cost reported by Composio.`;

app/src/providers/ChatRuntimeProvider.tsx

@@ -9,6 +9,7 @@
   type ChatSubagentSpawnedEvent,
   type ChatToolCallEvent,
   type ChatToolResultEvent,
+  type ProactiveMessageEvent,
   segmentText,
   subscribeChatEvents,
 } from '../services/chatService';
@@ -27,7 +28,12 @@
 } from '../store/chatRuntimeSlice';
 import { useAppDispatch, useAppSelector } from '../store/hooks';
 import { selectSocketStatus } from '../store/socketSelectors';
-import { addInferenceResponse, setActiveThread } from '../store/threadSlice';
+import {
+  addInferenceResponse,
+  createNewThread,
+  setActiveThread,
+  setSelectedThread,
+} from '../store/threadSlice';
 
 const logChatRuntime = debug('openhuman:chat-runtime');
 
@@ -55,6 +61,8 @@
   );
 
   const seenChatEventsRef = useRef<Map<string, number>>(new Map());
+  const proactiveThreadCreationPromiseRef = useRef<Promise<string | null> | null>(null);
+  const proactiveDispatchQueueRef = useRef<Promise<void>>(Promise.resolve());
   const toolTimelineRef = useRef(toolTimelineByThread);
   const inferenceStatusRef = useRef(inferenceStatusByThread);
   const streamingAssistantRef = useRef(streamingAssistantByThread);
@@ -104,6 +112,57 @@
     return true;
   };
 
+  const proactiveMessageDigest = (input: string): string => {
+    // Small non-cryptographic digest to keep dedupe keys bounded.
+    let hash = 2166136261;
+    for (let i = 0; i < input.length; i += 1) {
+      hash ^= input.charCodeAt(i);
+      hash = Math.imul(hash, 16777619);
+    }
+    return (hash >>> 0).toString(36);
+  };
+
+  const resolveVisibleThreadForProactive = async (
+    incomingThreadId: string
+  ): Promise<string | null> => {
+    if (!incomingThreadId.startsWith('proactive:')) {
+      return incomingThreadId;
+    }
+
+    const state = store.getState().thread;
+    const targetFromState =
+      state.selectedThreadId ?? state.activeThreadId ?? state.threads[0]?.id ?? null;
+    if (targetFromState) {
+      return targetFromState;
+    }
+
+    if (proactiveThreadCreationPromiseRef.current) {
+      return proactiveThreadCreationPromiseRef.current;
+    }
+
+    const createPromise: Promise<string | null> = (async () => {
+      try {
+        const newThread = await dispatch(createNewThread()).unwrap();
+        dispatch(setSelectedThread(newThread.id));
+        return newThread.id;
+      } catch (error) {
+        rtLog('proactive_thread_create_failed', {
+          err: error instanceof Error ? error.message : String(error),
+        });
+        return null;
+      } finally {
+        proactiveThreadCreationPromiseRef.current = null;
+      }
+    })();
+    proactiveThreadCreationPromiseRef.current = createPromise;
+
+    try {
+      return await createPromise;
+    } finally {
+      // no-op: cleared in createPromise.finally
+    }
+  };
+
   useEffect(() => {
     if (socketStatus !== 'connected') return;
 
@@ -371,6 +430,35 @@
         }
         dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries }));
       },
+      onProactiveMessage: (event: ProactiveMessageEvent) => {
+        const messageDigest = proactiveMessageDigest(event.full_response ?? '');
+        const eventKey = `proactive:${event.thread_id}:${event.request_id ?? 'none'}:${messageDigest}`;
+        if (
+          !markChatEventSeen(eventKey, { threadId: event.thread_id, requestId: event.request_id })
+        )
+          return;
+
+        proactiveDispatchQueueRef.current = proactiveDispatchQueueRef.current.then(async () => {
+          try {
+            const targetThreadId = await resolveVisibleThreadForProactive(event.thread_id);
+            if (!targetThreadId) return;
+            rtLog('proactive_message', {
+              from: event.thread_id,
+              to: targetThreadId,
+              request: event.request_id,
+            });
+            await dispatch(
+              addInferenceResponse({ content: event.full_response, threadId: targetThreadId })
+            );
+          } catch (error) {
+            rtLog('proactive_dispatch_failed', {
+              from: event.thread_id,
+              request: event.request_id,
+              error: error instanceof Error ? error.message : String(error),
+            });
+          }
+        });
+      },
       onDone: event => {
         const eventKey = `done:${event.thread_id}:${event.request_id ?? 'none'}`;
         if (
@@ -455,7 +543,7 @@
       rtLog('unsubscribe_chat_events');
       cleanup();
     };
   }, [dispatch, socketStatus]);
 
   return <>{children}</>;
 };

app/src/services/chatService.ts

@@ -82,6 +82,13 @@ export interface ChatErrorEvent {
   round: number | null;
 }
 
+/** Proactive assistant message pushed by the Rust event bus (not a chat turn). */
+export interface ProactiveMessageEvent {
+  thread_id: string;
+  request_id?: string;
+  full_response: string;
+}
+
 /** Emitted when the agent turn begins (before the first LLM call). */
 export interface ChatInferenceStartEvent {
   thread_id: string;
@@ -174,6 +181,7 @@ export interface ChatEventListeners {
   onTextDelta?: (event: ChatTextDeltaEvent) => void;
   onThinkingDelta?: (event: ChatThinkingDeltaEvent) => void;
   onToolArgsDelta?: (event: ChatToolArgsDeltaEvent) => void;
+  onProactiveMessage?: (event: ProactiveMessageEvent) => void;
   onDone?: (event: ChatDoneEvent) => void;
   onError?: (event: ChatErrorEvent) => void;
 }
@@ -198,6 +206,7 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
     textDelta: 'text_delta',
     thinkingDelta: 'thinking_delta',
     toolArgsDelta: 'tool_args_delta',
+    proactiveMessage: 'proactive_message',
     done: 'chat_done',
     error: 'chat_error',
   } as const;
@@ -383,6 +392,22 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
     handlers.push([EVENTS.toolArgsDelta, cb]);
   }
 
+  if (listeners.onProactiveMessage) {
+    const cb = (payload: unknown) => {
+      const e = payload as ProactiveMessageEvent;
+      chatLog(
+        '%s thread_id=%s request_id=%s chars=%d',
+        EVENTS.proactiveMessage,
+        e.thread_id,
+        e.request_id,
+        e.full_response?.length ?? 0
+      );
+      listeners.onProactiveMessage?.(e);
+    };
+    socket.on(EVENTS.proactiveMessage, cb);
+    handlers.push([EVENTS.proactiveMessage, cb]);
+  }
+
   if (listeners.onDone) {
     const cb = (payload: unknown) => {
       const e = payload as ChatDoneEvent;

src/openhuman/agent/agents/welcome/prompt.md

@@ -53,7 +53,7 @@ If `onboarding_status` is `"pending"`, continue the conversational flow below.
 Aim for this shape over **several** user/assistant turns — not one wall of text:
 
 1. **First substantive reply** — Concise greeting + what’s connected / not (from snapshot + `[CONNECTION_STATE]`) + one sentence on what OpenHuman is for (reasoning, memory, channels, integrations).
-2. **Gmail first** — Offer to help them connect Gmail. If they say yes, call `composio_authorize` with `{"toolkit": "gmail"}` and put the returned URL in a markdown link: `[Connect Gmail](url)`. Explicitly tell them clicking the link opens their default browser to complete auth.
+2. **Gmail first** — If the user's message already expresses intent to connect Gmail (e.g. "connect my Gmail", "give me the link", "I'd like to connect"), call `composio_authorize` with `{"toolkit": "gmail"}` **immediately** in that same response — no separate offer needed. Otherwise, offer first and wait for agreement. Either way, put the returned URL in a markdown link: `[Connect Gmail](url)` and explicitly tell them it opens in their default browser.
 3. **If they hesitate** — Once or twice, lightly explain why inbox access matters (triaging mail, drafts, etc.). **Do not** paste three auth links in a row or nag every line.
 4. **Try 2–3 times across the conversation** (not three demands in one message) to connect something. If they refuse everything, **wrap up kindly**: how to connect later in Settings, and that you’re here when they’re ready.
 5. **Show capability** — Weave examples into chat (e.g. “you could ask it to summarise yesterday’s mail”) instead of a bullet list brochure.
@@ -63,7 +63,7 @@ Aim for this shape over **several** user/assistant turns — not one wall of tex
 
 ## `composio_authorize` rules
 
-- Call **only after** the user agrees to connect that service.
+- Call when the user agrees to connect, or when their message already expresses clear intent to connect (e.g. "connect my Gmail", "give me the link"). No separate confirmation step needed in that case.
 - One toolkit at a time; Gmail is the default first offer.
 - Never invent URLs — only use `connectUrl` from the tool response, as a markdown link.
 - When sharing the link, clearly state it opens in the user's browser and they should return to chat after finishing auth.

src/openhuman/agent/welcome_proactive.rs

@@ -226,9 +226,14 @@ async fn run_proactive_welcome(config: Config) -> anyhow::Result<()> {
          have been exchanged yet. Write the personalised welcome message per your \
          system prompt. Do NOT call `complete_onboarding` with `complete` — the user \
          has not yet had a real conversation.\n\n\
+         Output format is STRICT JSON only:\n\
+         {{\"messages\":[\"<message 1>\",\"<message 2>\",\"<message 3>\"]}}\n\
+         - Return 2-4 assistant messages.\n\
+         - Each message should be short and conversational (1-3 sentences).\n\
+         - Do not include markdown code fences or any text outside the JSON object.\n\n\
          Status snapshot (treat exactly as if it were the tool return value):\n\
          ```json\n{snapshot_json}\n```\n\n\
-         Write your welcome message now."
+         Write your welcome messages now."
     );
     tracing::debug!(
         prompt_chars = prompt.len(),
@@ -258,31 +263,83 @@ async fn run_proactive_welcome(config: Config) -> anyhow::Result<()> {
         llm_result.map_err(|e| anyhow::anyhow!("welcome agent run_single failed: {e}"))?;
 
     let trimmed = response.trim();
+    // Tolerate common fenced JSON wrappers from the model:
+    // ```json
+    // { ... }
+    // ```
+    let trimmed = trimmed
+        .strip_prefix("```json")
+        .or_else(|| trimmed.strip_prefix("```"))
+        .map(str::trim_start)
+        .and_then(|s| s.strip_suffix("```"))
+        .map(str::trim)
+        .unwrap_or(trimmed);
     if trimmed.is_empty() {
         anyhow::bail!("welcome agent returned empty response");
     }
 
+    #[derive(serde::Deserialize)]
+    struct ProactiveMessagesPayload {
+        messages: Vec<String>,
+    }
+
+    // Preferred path: strict JSON payload from the model.
+    // Fallback path: split freeform prose into paragraph-ish chunks so users
+    // still receive multiple chat bubbles even if the provider drifts format.
+    let messages: Vec<String> = match serde_json::from_str::<ProactiveMessagesPayload>(trimmed) {
+        Ok(payload) => payload
+            .messages
+            .into_iter()
+            .map(|m| m.trim().to_string())
+            .filter(|m| !m.is_empty())
+            .collect(),
+        Err(parse_err) => {
+            tracing::warn!(
+                error = %parse_err,
+                "[welcome::proactive] response was not valid JSON payload; falling back to paragraph splitting"
+            );
+            trimmed
+                .split("\n\n")
+                .map(str::trim)
+                .filter(|s| !s.is_empty())
+                .map(ToOwned::to_owned)
+                .collect()
+        }
+    };
+
+    if messages.is_empty() {
+        anyhow::bail!("welcome agent returned no publishable messages");
+    }
+
     tracing::info!(
+        message_count = messages.len(),
         response_chars = trimmed.chars().count(),
-        "[welcome::proactive] welcome agent produced message — publishing ProactiveMessageRequested"
+        "[welcome::proactive] welcome agent produced multi-message payload — publishing ProactiveMessageRequested"
     );
 
-    // --- Publish LLM response (message 3) -----------------------------
-    publish_global(DomainEvent::ProactiveMessageRequested {
-        source: PROACTIVE_WELCOME_SOURCE.to_string(),
-        message: trimmed.to_string(),
-        job_name: Some(PROACTIVE_WELCOME_JOB_NAME.to_string()),
-    });
+    // --- Publish LLM responses (messages 3+) --------------------------
+    for (idx, message) in messages.iter().enumerate() {
+        if idx > 0 {
+            // Slight pacing so bubbles appear progressively instead of as a wall.
+            let pace_ms = (message.chars().count() as u64).clamp(600, 1200);
+            tokio::time::sleep(tokio::time::Duration::from_millis(pace_ms)).await;
+        }
+        publish_global(DomainEvent::ProactiveMessageRequested {
+            source: PROACTIVE_WELCOME_SOURCE.to_string(),
+            message: message.clone(),
+            job_name: Some(PROACTIVE_WELCOME_JOB_NAME.to_string()),
+        });
+    }
 
     // Post-publish confirmation so the log clearly marks end-to-end
     // success (publish_global is fire-and-forget; without this line,
     // "reached the end cleanly" is ambiguous from the log alone).
     tracing::debug!(
         source = PROACTIVE_WELCOME_SOURCE,
         job_name = PROACTIVE_WELCOME_JOB_NAME,
-        response_chars = trimmed.chars().count(),
+        published_llm_messages = messages.len(),
         "[welcome::proactive] proactive welcome flow complete \
-         (3 messages published: greeting, loading, llm-response)"
+         (template messages + multi-part llm response published)"
     );
 
     Ok(())