From 8cbb425cea40432a8a2224e2d62b3a98bc087011 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Mon, 13 Apr 2026 19:43:24 -0700 Subject: [PATCH] feat(onboarding): fire welcome agent immediately on completion (#548) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Implement proactive welcome agent for onboarding experience - Added a new module `welcome_proactive` to handle immediate welcome messages after user onboarding completion, enhancing user engagement. - Updated `set_onboarding_completed` to trigger the proactive welcome agent upon onboarding status change, ensuring timely delivery of welcome messages. - Refactored the onboarding process to streamline the integration of proactive messaging, improving overall user experience. - Enhanced documentation and logging for better observability of the onboarding flow and proactive message handling. - Introduced tests to validate the functionality of the new welcome agent and its integration into the onboarding process. * fix(socketio): auto-join "system" room so proactive messages reach clients The proactive message subscriber emits with `client_id="system"`, but connected Socket.IO clients only auto-joined a room named after their own sid — so the welcome agent's message was published into an empty room and never reached any frontend. Adds `socket.join("system")` at connect time alongside the existing per-sid join, so every client now receives broadcast proactive events (welcome agent, morning briefing, cron announcements). Also adds scripts/test-proactive-welcome.sh — end-to-end smoke test that resets onboarding flags, spawns a fresh core on port 7789, connects a Socket.IO listener, fires the RPC, and reports pass/miss for every checkpoint in the pipeline including client-side delivery. * fix(welcome): PR review — tighter gating, legacy-cron prune, safer harness - Gate proactive welcome spawn on `!was_chat_completed` so a user whose chat flow already completed (legacy tool path or manual flip) doesn't get a second welcome. - Prune legacy `welcome` cron jobs in `seed_proactive_agents`: one- shot entries left behind by an interrupted earlier run would otherwise double-deliver once the scheduler picks them up. Added a unit test covering the prune + morning-briefing seed in the same call. - Add a post-publish debug log in `welcome_proactive::run_proactive_welcome` so "reached end successfully" is distinguishable from "silently bailed above" by reading the log alone. - Replace ignored `socket.join(...)` results with a helper that logs success + failure per room + client id. Silent join failures on the "system" room make proactive delivery vanish without a trace. - Harden the test harness: * Back up CONFIG_PATH before mutating and restore on exit so the developer's staging profile is never permanently changed (unless `--keep-flags` is passed). * Tolerate inline comments + trailing whitespace in the TOML rewrite; append-at-end instead of prepend when a key is missing, so the rewrite can't accidentally land inside a section header. Not applied: - Unit tests for `run_proactive_welcome` empty-response / serialization- failure paths: empty response is already `anyhow::bail!`-guarded, and the serde_json failure branch is unreachable given a `json!`-built value. Testing either requires mocking `Agent::from_config_for_agent` which has heavy registry/provider dependencies — cost > value. --- scripts/test-proactive-welcome.sh | 214 ++++++++++++++++++ src/core/socketio.rs | 26 ++- src/openhuman/agent/mod.rs | 1 + src/openhuman/agent/welcome_proactive.rs | 173 ++++++++++++++ src/openhuman/config/ops.rs | 56 ++++- src/openhuman/cron/seed.rs | 168 ++++++++++---- .../tools/impl/agent/complete_onboarding.rs | 155 +++++++++---- src/openhuman/tools/impl/agent/mod.rs | 2 +- src/openhuman/tools/mod.rs | 2 +- 9 files changed, 697 insertions(+), 100 deletions(-) create mode 100755 scripts/test-proactive-welcome.sh create mode 100644 src/openhuman/agent/welcome_proactive.rs diff --git a/scripts/test-proactive-welcome.sh b/scripts/test-proactive-welcome.sh new file mode 100755 index 000000000..a11498366 --- /dev/null +++ b/scripts/test-proactive-welcome.sh @@ -0,0 +1,214 @@ +#!/usr/bin/env bash +# +# End-to-end smoke test for the proactive welcome flow. +# +# 1. Resets `onboarding_completed` + `chat_onboarding_completed` to false +# in the staging user's config.toml (the path a source-built binary reads). +# 2. Spawns a fresh `openhuman-core` binary on port 7789 with debug logs +# (non-default port so it doesn't fight a running `tauri dev` on 7788). +# 3. Connects a Socket.IO client that logs every event it receives. +# 4. Calls `openhuman.config_set_onboarding_completed` with value=true. +# 5. Watches the log up to 120s for each checkpoint in the pipeline. +# 6. Reports pass/miss per checkpoint AND whether the socket client got +# a `proactive_message` event. +# +# Usage: bash scripts/test-proactive-welcome.sh [--keep-flags] + +set -uo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +BIN="$REPO_ROOT/target/debug/openhuman-core" +PORT=7789 +USER_ID="69d9cb73e61f755583c3671f" +# Source-built binaries default to `.openhuman-staging`. Production +# staged binary reads `.openhuman`. We point at staging here. +CONFIG_ROOT="${OPENHUMAN_CONFIG_ROOT:-$HOME/.openhuman-staging}" +CONFIG_PATH="$CONFIG_ROOT/users/$USER_ID/config.toml" +LOG_FILE="$(mktemp -t openhuman-proactive-welcome-XXXXXX).log" +SIO_LOG="$(mktemp -t openhuman-sio-XXXXXX).log" +SIO_CLIENT_DIR="/tmp/sio-test" +KEEP_FLAGS=0 + +for arg in "$@"; do + case "$arg" in + --keep-flags) KEEP_FLAGS=1 ;; + esac +done + +log() { printf "[test] %s\n" "$*"; } +fail() { printf "[test][FAIL] %s\n" "$*" >&2; exit 1; } + +[[ -f "$BIN" ]] || fail "binary not built: $BIN (run: cargo build --bin openhuman-core)" +[[ -f "$CONFIG_PATH" ]] || fail "config not found: $CONFIG_PATH" + +# Flip the two onboarding keys to `false` in place, preserving any +# trailing inline comment and whitespace. If a key is missing, append +# a single line at the end of the file — never prepend, because the +# first line is usually a bare top-level assignment like +# `default_model = "..."` and prepending could land inside a section +# header on files laid out differently. +reset_flags() { + python3 - "$CONFIG_PATH" <<'PY' +import sys, re, pathlib +p = pathlib.Path(sys.argv[1]) +text = p.read_text() +# Match: start-of-line, key, optional spaces, =, spaces, true|false, +# optional trailing whitespace + "# comment" (captured so we can keep it). +for key in ("onboarding_completed", "chat_onboarding_completed"): + pat = re.compile( + rf'^(?P[ \t]*){key}[ \t]*=[ \t]*(?:true|false)(?P[ \t]*(?:#.*)?)$', + re.M, + ) + m = pat.search(text) + if m: + text = pat.sub(lambda mm: f"{mm.group('indent')}{key} = false{mm.group('tail')}", text, count=1) + else: + if not text.endswith("\n"): + text += "\n" + text += f"{key} = false\n" +p.write_text(text) +PY +} + +# Back up the config before touching it so cleanup can restore the +# user's original state verbatim (including comments, section order, +# and any unrelated fields). Belt-and-suspenders: we still call +# `reset_flags` pre-run to guarantee the two flags are `false` when +# the binary reads them, but the exit-trap uses `mv` of the backup +# so nothing we write survives unless `--keep-flags` is set. +CONFIG_BACKUP="${CONFIG_PATH}.bak.$$" +log "backing up $CONFIG_PATH -> $CONFIG_BACKUP" +cp "$CONFIG_PATH" "$CONFIG_BACKUP" + +log "resetting flags in $CONFIG_PATH" +reset_flags +grep -E '^(onboarding_completed|chat_onboarding_completed)\s*=' "$CONFIG_PATH" | sed 's/^/[test][config-before] /' + +log "starting $BIN on port $PORT (log: $LOG_FILE)" +RUST_LOG=debug,hyper=info,tungstenite=info,socketioxide=info \ + "$BIN" run --port "$PORT" > "$LOG_FILE" 2>&1 & +BIN_PID=$! + +cleanup() { + log "cleanup: killing bin pid=$BIN_PID (+ sio pid=${SIO_PID:-none})" + [[ -n "${SIO_PID:-}" ]] && kill "$SIO_PID" 2>/dev/null || true + if kill -0 "$BIN_PID" 2>/dev/null; then + kill "$BIN_PID" 2>/dev/null || true + wait "$BIN_PID" 2>/dev/null || true + fi + # Restore the original config from the backup — runs on both + # success and failure so the developer's staging profile is never + # permanently mutated by a test run. `--keep-flags` opts out so + # the flipped-to-true state survives for interactive debugging. + if [[ -f "$CONFIG_BACKUP" ]]; then + if [[ "$KEEP_FLAGS" -eq 0 ]]; then + log "restoring original config from $CONFIG_BACKUP" + mv "$CONFIG_BACKUP" "$CONFIG_PATH" + else + log "--keep-flags set; leaving backup at $CONFIG_BACKUP and current flag state in place" + fi + fi + log "binary log: $LOG_FILE" + log "socket log: $SIO_LOG" +} +trap cleanup EXIT + +log "waiting for core to be ready…" +for _ in $(seq 1 60); do + grep -q "OpenHuman core is ready" "$LOG_FILE" 2>/dev/null && break + sleep 0.5 +done +grep -q "OpenHuman core is ready" "$LOG_FILE" || { + tail -40 "$LOG_FILE" | sed 's/^/[test][core-log] /' + fail "core did not become ready" +} +log "core ready" + +# Give registry a moment. +sleep 1 + +# Spawn Socket.IO listener. +if [[ -f "$SIO_CLIENT_DIR/listen.js" && -d "$SIO_CLIENT_DIR/node_modules/socket.io-client" ]]; then + log "spawning socket.io listener -> $SIO_LOG" + (cd "$SIO_CLIENT_DIR" && node listen.js "$PORT" "$SIO_LOG" 150) > /dev/null 2>&1 & + SIO_PID=$! + sleep 2 + if grep -q CONNECTED "$SIO_LOG" 2>/dev/null; then + log "socket.io: $(grep CONNECTED "$SIO_LOG" | head -1)" + else + log "socket.io client did not confirm CONNECT; continuing anyway" + fi +else + log "socket.io-client not installed at $SIO_CLIENT_DIR — skipping" + SIO_PID="" +fi + +log "POST /rpc openhuman.config_set_onboarding_completed {value:true}" +RPC_RESP=$(curl -s -X POST "http://127.0.0.1:$PORT/rpc" \ + -H 'content-type: application/json' \ + -d '{"jsonrpc":"2.0","id":1,"method":"openhuman.config_set_onboarding_completed","params":{"value":true}}') +echo "[test][rpc-response] $RPC_RESP" +echo "$RPC_RESP" | grep -q '"result"' || fail "RPC did not return a result" + +log "watching log for welcome pipeline (timeout 120s)…" +CHECK_TRANSITION="[onboarding] false→true transition detected" +CHECK_SPAWN="[welcome::proactive] starting proactive welcome" +CHECK_INVOKE="[welcome::proactive] invoking welcome agent run_single" +CHECK_PRODUCED="[welcome::proactive] welcome agent produced message" +CHECK_PUBLISHED="[proactive] handling proactive message" +CHECK_EMITTED="[socketio] send event=proactive_message" + +deadline=$((SECONDS + 120)) +while (( SECONDS < deadline )); do + if grep -qF "$CHECK_PRODUCED" "$LOG_FILE" 2>/dev/null \ + || grep -qE "\[welcome::proactive\] failed to deliver" "$LOG_FILE" 2>/dev/null; then + break + fi + sleep 1 +done + +log "=== checkpoint summary (backend) ===" +for label in \ + "TRANSITION:$CHECK_TRANSITION" \ + "SPAWN:$CHECK_SPAWN" \ + "INVOKE:$CHECK_INVOKE" \ + "PRODUCED:$CHECK_PRODUCED" \ + "PUBLISHED:$CHECK_PUBLISHED" \ + "EMITTED:$CHECK_EMITTED"; do + name="${label%%:*}" + needle="${label#*:}" + if grep -qF "$needle" "$LOG_FILE" 2>/dev/null; then + printf "[test][PASS] %-11s %s\n" "$name" "$needle" + else + printf "[test][MISS] %-11s %s\n" "$name" "$needle" + fi +done + +# Wait a couple more seconds for the socket event round-trip, then inspect. +sleep 3 +log "=== client-side socket.io events ===" +if [[ -f "$SIO_LOG" && -s "$SIO_LOG" ]]; then + cat "$SIO_LOG" | sed 's/^/[test][sio] /' + if grep -q 'EVENT proactive_message' "$SIO_LOG" 2>/dev/null \ + || grep -q 'EVENT proactive:message' "$SIO_LOG" 2>/dev/null; then + printf "[test][PASS] %-11s %s\n" "DELIVERY" "socket.io client received proactive_message" + else + printf "[test][MISS] %-11s %s\n" "DELIVERY" "socket.io client did NOT receive proactive_message (server emitted to room=system; clients auto-join only their own sid room)" + fi +else + log "no socket.io log (listener not started)" +fi + +log "=== welcome agent full message (from log) ===" +python3 - "$LOG_FILE" <<'PY' +import re, pathlib, sys +t = pathlib.Path(sys.argv[1]).read_text() +m = re.search(r'provider response: ChatResponse \{ text: Some\("(.*?)"\)', t, re.S) +if m: + body = m.group(1).encode('utf-8').decode('unicode_escape') + print(body) +else: + print("(no final assistant text found in log)") +PY + +echo "[test] done." diff --git a/src/core/socketio.rs b/src/core/socketio.rs index da9820ea0..419538e6a 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -101,7 +101,16 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) { let client_id = socket.id.to_string(); log::info!("[socketio] client connected id={client_id}"); // Join a room named after the client ID for targeted event delivery. - let _ = socket.join(client_id.clone()); + join_room_logged(&socket, &client_id, &client_id); + // Also auto-join the "system" room so every connected client + // receives broadcast-style events that aren't tied to a + // specific chat thread. Today this covers proactive messages + // (welcome agent, morning briefing, cron-driven announcements) + // which `channels::proactive::ProactiveMessageSubscriber` + // emits with `client_id = "system"` — see `emit_web_channel_event`. + // If this join fails the welcome message silently disappears, + // so we log both success and failure for diagnosability. + join_room_logged(&socket, "system", &client_id); let ready_payload = json!({ "sid": client_id }); log::debug!("[socketio] emit event=ready to_client={}", socket.id); let _ = socket.emit("ready", &ready_payload); @@ -327,6 +336,21 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { }); } +/// Join `socket` to `room`, logging the result. +/// +/// `socket.join()` returns a `Result` that historically was discarded +/// with `let _ = …`. Silent failure on the `"system"` room in +/// particular makes proactive-message delivery vanish without a trace, +/// so both the happy and error paths are logged with enough context +/// (room name + client id) to diagnose missing welcome messages from +/// logs alone. +fn join_room_logged(socket: &SocketRef, room: &str, client_id: &str) { + match socket.join(room.to_string()) { + Ok(()) => log::debug!("[socketio] joined room '{room}' for client {client_id}"), + Err(e) => log::warn!("[socketio] failed to join room '{room}' for client {client_id}: {e}"), + } +} + fn emit_web_channel_event(io: &SocketIo, event: WebChannelEvent) { let room = event.client_id.clone(); let name = event.event.clone(); diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index 57ada78de..e30c1a5a7 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -31,6 +31,7 @@ pub mod pformat; pub mod progress; mod schemas; pub mod triage; +pub mod welcome_proactive; pub use schemas::{ all_controller_schemas as all_agent_controller_schemas, all_registered_controllers as all_agent_registered_controllers, diff --git a/src/openhuman/agent/welcome_proactive.rs b/src/openhuman/agent/welcome_proactive.rs new file mode 100644 index 000000000..778d61008 --- /dev/null +++ b/src/openhuman/agent/welcome_proactive.rs @@ -0,0 +1,173 @@ +//! Proactive welcome — fires the welcome agent immediately when the +//! user completes the desktop onboarding wizard, instead of waiting +//! for their first chat message. +//! +//! ## Flow +//! +//! 1. [`crate::openhuman::config::ops::set_onboarding_completed`] +//! detects a false→true transition and calls [`spawn_proactive_welcome`]. +//! 2. That function spawns a detached Tokio task that: +//! - Builds the same JSON status snapshot +//! `tools::implementations::agent::complete_onboarding`'s +//! `check_status` would have returned, with `finalize_action = +//! "flipped"` (the caller has already flipped +//! `chat_onboarding_completed`). +//! - Loads the `welcome` agent via +//! [`crate::openhuman::agent::Agent::from_config_for_agent`] so +//! the agent runs with its own `prompt.md`, tool allowlist, and +//! model hint. +//! - Calls [`crate::openhuman::agent::Agent::run_single`] with a +//! prompt that embeds the snapshot and instructs the agent to +//! skip iteration 1 (the tool call) and go straight to iteration +//! 2 (writing the welcome message). Because we already flipped +//! `chat_onboarding_completed`, the `complete_onboarding` tool +//! would be a no-op anyway — but avoiding the extra round-trip +//! keeps latency down. +//! - On success, publishes +//! [`DomainEvent::ProactiveMessageRequested`] so the existing +//! [`crate::openhuman::channels::proactive::ProactiveMessageSubscriber`] +//! delivers the message to the web channel (and any active +//! external channel) without any new transport code. +//! +//! All steps log at `debug` / `info` so operators can trace the +//! proactive welcome end-to-end: `[welcome::proactive] ...`. + +use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::openhuman::agent::Agent; +use crate::openhuman::config::Config; +use crate::openhuman::tools::implementations::agent::complete_onboarding::build_status_snapshot; + +/// Event-bus `source` label attached to the proactive welcome message. +/// Kept as a constant so tests and channel-side filters have a stable +/// grep target. +pub const PROACTIVE_WELCOME_SOURCE: &str = "onboarding_completed"; + +/// Job name used when publishing [`DomainEvent::ProactiveMessageRequested`]. +/// Matches the cron-job naming convention so +/// [`crate::openhuman::channels::proactive::ProactiveMessageSubscriber`] +/// routes it under `proactive:welcome`. +pub const PROACTIVE_WELCOME_JOB_NAME: &str = "welcome"; + +/// Fire-and-forget launch of the welcome agent after onboarding +/// completes. +/// +/// Spawned on a detached Tokio task so the caller's RPC response +/// path is never blocked. Failures are logged at `warn` and +/// swallowed — the welcome is best-effort, and the user can still +/// get a (less-polished) welcome by sending their first message +/// (which would route through the normal dispatch path, since the +/// caller flips `chat_onboarding_completed` before invoking us). +pub fn spawn_proactive_welcome(config: Config) { + tokio::spawn(async move { + if let Err(e) = run_proactive_welcome(config).await { + tracing::warn!( + error = %e, + "[welcome::proactive] failed to deliver proactive welcome — \ + falling back to on-first-message flow" + ); + } + }); +} + +/// Internal: build the snapshot, run the welcome agent, publish the +/// result. Split out from the spawn so it can be unit-tested with +/// an injected Config + mocked provider. +async fn run_proactive_welcome(config: Config) -> anyhow::Result<()> { + tracing::info!( + "[welcome::proactive] starting proactive welcome (chat_onboarding_completed={}, ui_onboarding_completed={})", + config.chat_onboarding_completed, + config.onboarding_completed + ); + + // The caller (set_onboarding_completed) already flipped + // `chat_onboarding_completed`, so the snapshot always reports + // `"flipped"` — matches the state the welcome agent's prompt.md + // treats as "first run, authenticated, deliver the full welcome". + let snapshot = build_status_snapshot(&config, "flipped"); + let snapshot_json = serde_json::to_string_pretty(&snapshot) + .map_err(|e| anyhow::anyhow!("serialize status snapshot: {e}"))?; + tracing::debug!( + snapshot_chars = snapshot_json.len(), + "[welcome::proactive] built status snapshot" + ); + + let mut agent = Agent::from_config_for_agent(&config, "welcome").map_err(|e| { + anyhow::anyhow!("build welcome agent: {e} — ensure AgentDefinitionRegistry is initialised") + })?; + agent.set_event_context( + format!("proactive:{PROACTIVE_WELCOME_JOB_NAME}"), + "proactive", + ); + + // The welcome prompt.md is insistent that iteration 1 must be a + // `complete_onboarding` tool call. We bypass that here by + // pre-delivering the snapshot inside the user message and + // explicitly overriding iteration 1. Capable models comply; if a + // model calls the tool anyway, the tool returns + // `finalize_action: "already_complete"` (we've pre-flipped the + // flag) so the message still lands — just with a slightly + // different framing. + let prompt = format!( + "[PROACTIVE INVOCATION — the user just finished the desktop onboarding wizard; \ + this is not a reply to anything they typed, it is your opening message.]\n\n\ + Skip iteration 1. Do NOT call `complete_onboarding` or any other tool. The \ + status snapshot that `complete_onboarding(check_status)` would have returned \ + is already provided below. Jump straight to iteration 2 and write the \ + personalised welcome message per your system prompt guidelines.\n\n\ + Status snapshot (treat exactly as if it were the tool return value):\n\ + ```json\n{snapshot_json}\n```\n\n\ + Write iteration 2 now." + ); + tracing::debug!( + prompt_chars = prompt.len(), + "[welcome::proactive] invoking welcome agent run_single" + ); + + let response = agent + .run_single(&prompt) + .await + .map_err(|e| anyhow::anyhow!("welcome agent run_single failed: {e}"))?; + + let trimmed = response.trim(); + if trimmed.is_empty() { + anyhow::bail!("welcome agent returned empty response"); + } + + tracing::info!( + response_chars = trimmed.chars().count(), + "[welcome::proactive] welcome agent produced message — publishing ProactiveMessageRequested" + ); + + publish_global(DomainEvent::ProactiveMessageRequested { + source: PROACTIVE_WELCOME_SOURCE.to_string(), + message: trimmed.to_string(), + job_name: Some(PROACTIVE_WELCOME_JOB_NAME.to_string()), + }); + + // Post-publish confirmation. `publish_global` is a best-effort + // broadcast send that swallows lag / no-subscriber errors, so + // without this line the caller can't distinguish "reached the + // end successfully" from "silently bailed somewhere above" by + // reading the log alone. + tracing::debug!( + source = PROACTIVE_WELCOME_SOURCE, + job_name = PROACTIVE_WELCOME_JOB_NAME, + response_chars = trimmed.chars().count(), + "[welcome::proactive] proactive welcome flow complete" + ); + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn source_and_job_name_constants_are_stable() { + // These strings show up in channel-side filters and logs — a + // silent rename would break downstream grep-based traces. + assert_eq!(PROACTIVE_WELCOME_SOURCE, "onboarding_completed"); + assert_eq!(PROACTIVE_WELCOME_JOB_NAME, "welcome"); + } +} diff --git a/src/openhuman/config/ops.rs b/src/openhuman/config/ops.rs index e8237a83b..faf904a57 100644 --- a/src/openhuman/config/ops.rs +++ b/src/openhuman/config/ops.rs @@ -569,32 +569,72 @@ pub async fn get_onboarding_completed() -> Result, String> { /// Updates and persists the onboarding completion status. /// -/// When transitioning to `true`, seeds default proactive agent cron jobs -/// (morning briefing + one-shot welcome). Seeding is fire-and-forget so -/// it never blocks the RPC response. +/// On a false→true transition this does three things before returning: +/// +/// 1. Sets `chat_onboarding_completed = true` on the same config save, +/// so the user's next chat message routes straight to the +/// orchestrator rather than the welcome agent (which is about to +/// send its message proactively below). See +/// [`crate::openhuman::channels::runtime::dispatch::resolve_target_agent`] +/// for the routing contract. +/// +/// 2. Seeds the recurring morning-briefing cron job via +/// [`crate::openhuman::cron::seed::seed_proactive_agents`]. +/// +/// 3. Spawns the welcome agent immediately via +/// [`crate::openhuman::agent::welcome_proactive::spawn_proactive_welcome`], +/// so the first welcome message arrives the moment the user +/// finishes the wizard instead of waiting for them to type. +/// +/// All three side-effects are fire-and-forget so the RPC response +/// lands before any agent work completes. pub async fn set_onboarding_completed(value: bool) -> Result, String> { tracing::debug!(value, "[onboarding] set_onboarding_completed called"); let mut config = load_config_with_timeout().await?; let was_completed = config.onboarding_completed; config.onboarding_completed = value; + + // On the false→true transition, also flip the chat-side flag so + // the welcome agent we're about to invoke proactively is the + // *only* place the user hears the welcome copy — their first + // typed message routes straight to the orchestrator. + let was_chat_completed = config.chat_onboarding_completed; + if value && !was_completed && !was_chat_completed { + tracing::debug!( + "[onboarding] flipping chat_onboarding_completed=true alongside ui onboarding flag" + ); + config.chat_onboarding_completed = true; + } + config.save().await.map_err(|e| e.to_string())?; - // Seed proactive agents exactly once, on the false→true transition. + // Seed proactive agents (morning briefing) on any UI false→true + // transition. The welcome agent fires only when the *chat* flow + // hasn't completed yet — otherwise a user whose chat welcome was + // already delivered (e.g. via the legacy tool path, or a manual + // flip) would get a second welcome. if value && !was_completed { - tracing::debug!( - "[onboarding] false→true transition detected, spawning proactive agent seeding" - ); + tracing::debug!("[onboarding] false→true transition detected — seeding morning briefing"); let seed_config = config.clone(); tokio::task::spawn_blocking(move || { if let Err(e) = crate::openhuman::cron::seed::seed_proactive_agents(&seed_config) { tracing::warn!("[onboarding] failed to seed proactive agent cron jobs: {e}"); } }); + + if !was_chat_completed { + tracing::debug!("[onboarding] chat flow not yet completed — firing proactive welcome"); + crate::openhuman::agent::welcome_proactive::spawn_proactive_welcome(config.clone()); + } else { + tracing::debug!( + "[onboarding] chat_onboarding_completed already true — skipping proactive welcome" + ); + } } else { tracing::debug!( was_completed, value, - "[onboarding] no transition — skipping proactive agent seeding" + "[onboarding] no transition — skipping proactive seeding and welcome" ); } diff --git a/src/openhuman/cron/seed.rs b/src/openhuman/cron/seed.rs index 227339ba2..2fb6d19ef 100644 --- a/src/openhuman/cron/seed.rs +++ b/src/openhuman/cron/seed.rs @@ -2,21 +2,34 @@ //! //! Called once after onboarding completes to create: //! - A recurring daily morning briefing job (7 AM, user's local time or UTC) -//! - A one-shot welcome agent job that fires within seconds //! -//! Both use `mode: "proactive"` delivery so the channels module's -//! [`ProactiveMessageSubscriber`] routes to the user's active channel. +//! The morning briefing uses `mode: "proactive"` delivery so the +//! channels module's +//! [`crate::openhuman::channels::proactive::ProactiveMessageSubscriber`] +//! routes to the user's active channel. +//! +//! The one-shot welcome message used to be seeded here too, but it is +//! now fired immediately by +//! [`crate::openhuman::agent::welcome_proactive::spawn_proactive_welcome`] +//! from `config::ops::set_onboarding_completed` — no cron round-trip +//! needed. Users who seeded the legacy welcome job under a prior build +//! have any stale entry pruned here (see [`prune_legacy_welcome`]) so +//! the scheduler can't double-deliver. use crate::openhuman::config::Config; use crate::openhuman::cron::{ - add_agent_job_with_definition, list_jobs, DeliveryConfig, Schedule, SessionTarget, + add_agent_job_with_definition, list_jobs, remove_job, DeliveryConfig, Schedule, SessionTarget, }; use anyhow::Result; -use chrono::{Duration, Utc}; /// Well-known job names used to detect whether seeding has already run. const MORNING_BRIEFING_JOB_NAME: &str = "morning_briefing"; -const WELCOME_JOB_NAME: &str = "welcome"; + +/// Legacy name of the one-shot welcome cron job created by earlier +/// builds of `seed_proactive_agents`. Kept as a constant (rather than +/// a string literal inline) so a grep for `WELCOME_JOB_NAME` still +/// finds the migration path. +const LEGACY_WELCOME_JOB_NAME: &str = "welcome"; /// Delivery config for proactive agents. The channels module decides /// which channel(s) to deliver to based on the user's active channel @@ -33,10 +46,16 @@ fn proactive_delivery() -> DeliveryConfig { /// Seed the proactive agent cron jobs after onboarding completes. /// /// Idempotent: skips creation if jobs with matching names already exist. +/// Also prunes any stale one-shot `welcome` job a prior build might +/// have persisted (see [`prune_legacy_welcome`]). pub fn seed_proactive_agents(config: &Config) -> Result<()> { let existing = list_jobs(config)?; let has = |name: &str| existing.iter().any(|j| j.name.as_deref() == Some(name)); + // Prune before re-listing so a legacy welcome job left over from + // an interrupted prior run can't deliver a second welcome. + prune_legacy_welcome(config, &existing); + if !has(MORNING_BRIEFING_JOB_NAME) { tracing::info!("[cron::seed] creating morning_briefing daily cron job"); seed_morning_briefing(config)?; @@ -44,14 +63,44 @@ pub fn seed_proactive_agents(config: &Config) -> Result<()> { tracing::debug!("[cron::seed] morning_briefing job already exists — skipping"); } - if !has(WELCOME_JOB_NAME) { - tracing::info!("[cron::seed] creating one-shot welcome agent job"); - seed_welcome(config)?; - } else { - tracing::debug!("[cron::seed] welcome job already exists — skipping"); + Ok(()) +} + +/// Remove any persisted cron job named `"welcome"` from a prior build. +/// +/// The one-shot welcome job `delete_after_run = true + Schedule::At` +/// self-cleans on success, but if the scheduler never got a chance to +/// fire it (upgrade mid-window, scheduler disabled, process killed +/// before the 10-second fire-at) the entry can persist. Now that the +/// welcome is delivered immediately by +/// [`crate::openhuman::agent::welcome_proactive::spawn_proactive_welcome`], +/// letting such an entry fire would double-deliver. Best-effort: log +/// but don't fail seeding on a prune error, and scan all entries +/// because the ID is a UUID — we key on the stable `name` field. +fn prune_legacy_welcome(config: &Config, existing: &[crate::openhuman::cron::CronJob]) { + let stale_ids: Vec = existing + .iter() + .filter(|j| j.name.as_deref() == Some(LEGACY_WELCOME_JOB_NAME)) + .map(|j| j.id.clone()) + .collect(); + + if stale_ids.is_empty() { + return; } - Ok(()) + tracing::info!( + count = stale_ids.len(), + "[cron::seed] pruning legacy '{LEGACY_WELCOME_JOB_NAME}' cron job(s) — welcome is now delivered immediately" + ); + for id in stale_ids { + if let Err(e) = remove_job(config, &id) { + tracing::warn!( + job_id = %id, + error = %e, + "[cron::seed] failed to remove legacy welcome cron job — continuing" + ); + } + } } /// Daily morning briefing at 7:00 AM UTC. @@ -86,48 +135,30 @@ fn seed_morning_briefing(config: &Config) -> Result<()> { Ok(()) } -/// One-shot welcome message that fires ~10 seconds after creation. -/// -/// Uses `Schedule::At` with a near-future timestamp so the scheduler -/// picks it up on its next poll cycle. `delete_after_run = true` ensures -/// the job is cleaned up after the welcome is delivered. -fn seed_welcome(config: &Config) -> Result<()> { - let fire_at = Utc::now() + Duration::seconds(10); - let schedule = Schedule::At { at: fire_at }; - - let prompt = concat!( - "You are the welcome agent. The user just finished setting up ", - "their OpenHuman workspace. Review everything you know about them ", - "from memory — connected integrations, profile details, onboarding ", - "choices — and deliver a witty, personalized welcome message. Be ", - "snarky but warm. Show that you already understand them. Keep it ", - "to 150-250 words." - ); - - add_agent_job_with_definition( - config, - Some(WELCOME_JOB_NAME.to_string()), - schedule, - prompt, - SessionTarget::Isolated, - None, - Some(proactive_delivery()), - true, // one-shot — delete after successful run - Some(WELCOME_JOB_NAME.to_string()), - )?; - - Ok(()) -} - #[cfg(test)] mod tests { use super::*; + use crate::openhuman::cron::{ + add_agent_job_with_definition, list_jobs, Schedule, SessionTarget, + }; + use chrono::{Duration as ChronoDuration, Utc}; + use tempfile::TempDir; + + fn test_config(tmp: &TempDir) -> Config { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + config + } #[test] fn constants_are_valid_identifiers() { assert!(!MORNING_BRIEFING_JOB_NAME.is_empty()); - assert!(!WELCOME_JOB_NAME.is_empty()); - assert_ne!(MORNING_BRIEFING_JOB_NAME, WELCOME_JOB_NAME); + assert!(!LEGACY_WELCOME_JOB_NAME.is_empty()); + assert_ne!(MORNING_BRIEFING_JOB_NAME, LEGACY_WELCOME_JOB_NAME); } #[test] @@ -138,4 +169,47 @@ mod tests { assert!(d.to.is_none()); assert!(d.best_effort); } + + #[test] + fn seed_prunes_legacy_welcome_job() { + // Simulate the state an earlier build would have left behind: + // a one-shot cron job named "welcome" that never fired + // (scheduler off, process killed before the 10-second + // window, etc.). seed_proactive_agents should delete it so + // the new immediate-fire welcome path doesn't double-deliver. + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let fire_at = Utc::now() + ChronoDuration::hours(1); + add_agent_job_with_definition( + &config, + Some(LEGACY_WELCOME_JOB_NAME.to_string()), + Schedule::At { at: fire_at }, + "legacy welcome prompt", + SessionTarget::Isolated, + None, + Some(proactive_delivery()), + true, + Some(LEGACY_WELCOME_JOB_NAME.to_string()), + ) + .expect("seed legacy welcome"); + assert_eq!(list_jobs(&config).unwrap().len(), 1); + + seed_proactive_agents(&config).expect("seed should succeed"); + + let remaining = list_jobs(&config).unwrap(); + assert!( + !remaining + .iter() + .any(|j| j.name.as_deref() == Some(LEGACY_WELCOME_JOB_NAME)), + "legacy welcome job should have been pruned, got: {remaining:?}" + ); + // Morning briefing should have been seeded in its place. + assert!( + remaining + .iter() + .any(|j| j.name.as_deref() == Some(MORNING_BRIEFING_JOB_NAME)), + "morning_briefing should have been seeded, got: {remaining:?}" + ); + } } diff --git a/src/openhuman/tools/impl/agent/complete_onboarding.rs b/src/openhuman/tools/impl/agent/complete_onboarding.rs index 6d8fdb9cf..58860597f 100644 --- a/src/openhuman/tools/impl/agent/complete_onboarding.rs +++ b/src/openhuman/tools/impl/agent/complete_onboarding.rs @@ -164,31 +164,7 @@ async fn check_status() -> anyhow::Result { .await .map_err(|e| anyhow::anyhow!("Failed to load config: {e}"))?; - // ── Auth detection ──────────────────────────────────────────── - // - // Two possible auth sources, in priority order: - // - // 1. The `app-session:default` profile in `auth-profiles.json` — - // the canonical inference credential, populated by the desktop - // OAuth deep-link flow's `exchange_token` Rust command. This is - // where every production inference RPC reads from. - // 2. `config.api_key` — legacy free-form provider key field, kept - // for CI / dev setups that bypass the desktop login flow. - // - // Either one counts as "authenticated" for the welcome flow. - let has_session_jwt = crate::api::jwt::get_session_token(&config) - .ok() - .flatten() - .is_some_and(|t| !t.is_empty()); - let has_legacy_api_key = config.api_key.as_ref().is_some_and(|k| !k.is_empty()); - let is_authenticated = has_session_jwt || has_legacy_api_key; - let auth_source: Value = if has_session_jwt { - Value::String("session_token".to_string()) - } else if has_legacy_api_key { - Value::String("legacy_api_key".to_string()) - } else { - Value::Null - }; + let (is_authenticated, _auth_source) = detect_auth(&config); // ── Auto-finalize side effect ───────────────────────────────── // @@ -221,6 +197,67 @@ async fn check_status() -> anyhow::Result { "flipped" }; + let snapshot = build_status_snapshot(&config, finalize_action); + let payload = serde_json::to_string_pretty(&snapshot) + .map_err(|e| anyhow::anyhow!("Failed to serialize status snapshot: {e}"))?; + + tracing::debug!( + "[complete_onboarding] check_status returned authenticated={} finalize_action={} chars={}", + is_authenticated, + finalize_action, + payload.len() + ); + + Ok(ToolResult::success(payload)) +} + +/// Detect whether the user is authenticated for the welcome flow. +/// +/// Two possible auth sources, in priority order: +/// +/// 1. The `app-session:default` profile in `auth-profiles.json` — +/// the canonical inference credential, populated by the desktop +/// OAuth deep-link flow's `exchange_token` Rust command. This is +/// where every production inference RPC reads from. +/// 2. `config.api_key` — legacy free-form provider key field, kept +/// for CI / dev setups that bypass the desktop login flow. +/// +/// Either one counts as "authenticated". +/// +/// Returned as `(is_authenticated, auth_source_json)` so callers can +/// both gate behaviour on the bool and embed the source label in a +/// JSON payload without rebuilding the logic. +pub(crate) fn detect_auth(config: &Config) -> (bool, Value) { + let has_session_jwt = crate::api::jwt::get_session_token(config) + .ok() + .flatten() + .is_some_and(|t| !t.is_empty()); + let has_legacy_api_key = config.api_key.as_ref().is_some_and(|k| !k.is_empty()); + let is_authenticated = has_session_jwt || has_legacy_api_key; + let auth_source: Value = if has_session_jwt { + Value::String("session_token".to_string()) + } else if has_legacy_api_key { + Value::String("legacy_api_key".to_string()) + } else { + Value::Null + }; + (is_authenticated, auth_source) +} + +/// Build the structured JSON snapshot that the welcome agent consumes. +/// +/// The snapshot describes the user's workspace setup (connected +/// channels, integrations, delegate agents, memory settings, and +/// onboarding flags) and embeds a `finalize_action` discriminator so +/// the agent knows which message framing to use. Shared between +/// [`check_status`] (reactive, called by the tool) and the proactive +/// welcome path (fired on `onboarding_completed` false→true). +/// +/// The caller is responsible for computing `finalize_action` — +/// typically `"flipped"`, `"already_complete"`, or `"skipped_no_auth"`. +pub(crate) fn build_status_snapshot(config: &Config, finalize_action: &str) -> Value { + let (is_authenticated, auth_source) = detect_auth(config); + // ── Connected messaging channels ────────────────────────────── let mut channels_connected: Vec<&str> = Vec::new(); if config.channels_config.telegram.is_some() { @@ -266,7 +303,6 @@ async fn check_status() -> anyhow::Result { channels_connected.push("qq"); } - // ── Integrations ────────────────────────────────────────────── let composio_enabled = config.composio.enabled && config .composio @@ -274,11 +310,9 @@ async fn check_status() -> anyhow::Result { .as_ref() .is_some_and(|k| !k.is_empty()); - // ── Delegate agents ─────────────────────────────────────────── let delegate_agents: Vec<&str> = config.agents.keys().map(|s| s.as_str()).collect(); - // ── Build the JSON snapshot ─────────────────────────────────── - let snapshot = json!({ + json!({ "authenticated": is_authenticated, "auth_source": auth_source, "default_model": config @@ -306,19 +340,7 @@ async fn check_status() -> anyhow::Result { "ui_onboarding_completed": config.onboarding_completed, "chat_onboarding_completed": config.chat_onboarding_completed, "finalize_action": finalize_action, - }); - - let payload = serde_json::to_string_pretty(&snapshot) - .map_err(|e| anyhow::anyhow!("Failed to serialize status snapshot: {e}"))?; - - tracing::debug!( - "[complete_onboarding] check_status returned authenticated={} finalize_action={} chars={}", - is_authenticated, - finalize_action, - payload.len() - ); - - Ok(ToolResult::success(payload)) + }) } /// Legacy manual finalize-only path. Flips `chat_onboarding_completed` @@ -371,4 +393,53 @@ mod tests { assert!(schema["properties"]["action"].is_object()); assert_eq!(schema["required"], serde_json::json!(["action"])); } + + #[test] + fn build_status_snapshot_carries_finalize_action_and_core_fields() { + // A default Config is "bare install" — no channels, no + // integrations. This test locks in the JSON shape the welcome + // agent's prompt.md depends on, and is the contract shared + // between `check_status` (reactive) and the proactive welcome + // path — if a future refactor drops or renames a field, this + // fails loudly. + let config = Config::default(); + let snapshot = build_status_snapshot(&config, "flipped"); + + assert_eq!(snapshot["finalize_action"], "flipped"); + assert_eq!(snapshot["chat_onboarding_completed"], false); + assert_eq!(snapshot["ui_onboarding_completed"], false); + assert_eq!(snapshot["active_channel"], "web"); + assert_eq!( + snapshot["channels_connected"] + .as_array() + .expect("channels_connected is an array") + .len(), + 0, + "default Config should report zero connected channels" + ); + assert!(snapshot["integrations"].is_object()); + assert!(snapshot["memory"].is_object()); + // Every integration flag present so the welcome prompt can + // branch on bare-install handling without optional-chain checks. + for key in [ + "composio", + "browser", + "web_search", + "http_request", + "local_ai", + ] { + assert!( + snapshot["integrations"][key].is_boolean(), + "integrations.{key} must be a bool" + ); + } + } + + #[test] + fn detect_auth_on_default_config_is_unauthenticated() { + let config = Config::default(); + let (is_auth, source) = detect_auth(&config); + assert!(!is_auth); + assert!(source.is_null()); + } } diff --git a/src/openhuman/tools/impl/agent/mod.rs b/src/openhuman/tools/impl/agent/mod.rs index ffea55e18..7388c3b82 100644 --- a/src/openhuman/tools/impl/agent/mod.rs +++ b/src/openhuman/tools/impl/agent/mod.rs @@ -1,6 +1,6 @@ mod archetype_delegation; mod ask_clarification; -mod complete_onboarding; +pub(crate) mod complete_onboarding; mod delegate; mod skill_delegation; mod spawn_subagent; diff --git a/src/openhuman/tools/mod.rs b/src/openhuman/tools/mod.rs index e2ff87601..bfe75f87d 100644 --- a/src/openhuman/tools/mod.rs +++ b/src/openhuman/tools/mod.rs @@ -6,7 +6,7 @@ mod schemas; pub mod traits; #[path = "impl/mod.rs"] -mod implementations; +pub(crate) mod implementations; pub use implementations::*; pub use ops::*;