From 64975b8a7ea0fc1be422b27e4a9cd083fef42579 Mon Sep 17 00:00:00 2001 From: Freddy Diaz <45578633+FreddyJD@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:05:22 -0400 Subject: [PATCH 1/2] feat(bots): replace loops with top-level bot chats and durable automation Add Facehash bot navigation, persistent roles, schedules, cross-session delegation, and main-process queue ownership. Migrate existing loops in place and cover bot runtime, IPC, and UI behavior. Co-authored-by: Roxy <299891354+roxy-commits@users.noreply.github.com> --- BOTS.md | 68 +++ README.md | 6 +- package-lock.json | 43 ++ package.json | 3 + src/main/db/bots.ts | 177 +++++++ src/main/db/migrations.ts | 100 +++- src/main/db/repo.ts | 213 +++----- src/main/harness/agent.ts | 123 ++++- src/main/harness/bot-tools.ts | 229 ++++++++ src/main/harness/tools.ts | 81 +-- src/main/index.ts | 7 +- src/main/ipc/index.ts | 126 ++++- src/main/services/automation.ts | 408 ++++++++++++++ src/main/services/compaction.ts | 2 +- src/main/services/loops.ts | 49 -- src/main/services/remote.ts | 324 +----------- src/main/services/session-turn.ts | 1 + src/main/services/turn-state.ts | 31 ++ src/main/services/workspace.ts | 8 +- src/preload/index.ts | 41 +- src/renderer/src/canvas/CanvasTranscript.tsx | 18 +- src/renderer/src/canvas/transcript-window.ts | 18 +- src/renderer/src/canvas/transcript.ts | 34 +- src/renderer/src/components/BotAvatar.tsx | 45 ++ .../src/components/BotSettingsPane.tsx | 499 ++++++++++++++++++ src/renderer/src/components/BotsSection.tsx | 165 ++++++ src/renderer/src/components/ChatView.tsx | 67 +-- src/renderer/src/components/Composer.tsx | 126 ++++- .../src/components/LoopDetailsPane.tsx | 137 ----- src/renderer/src/components/LoopsSection.tsx | 113 ---- src/renderer/src/components/QueuedMessage.tsx | 71 ++- src/renderer/src/components/Sidebar.tsx | 68 +-- src/renderer/src/lib/store.ts | 339 +++++++----- src/renderer/src/locales/ar.json | 77 ++- src/renderer/src/locales/de.json | 77 ++- src/renderer/src/locales/default.json | 75 ++- src/renderer/src/locales/es.json | 77 ++- src/renderer/src/locales/fr.json | 77 ++- src/renderer/src/locales/hi.json | 77 ++- src/renderer/src/locales/ja.json | 77 ++- src/renderer/src/locales/pt.json | 77 ++- src/renderer/src/locales/ru.json | 77 ++- src/renderer/src/locales/zh.json | 77 ++- src/shared/api.ts | 37 +- src/shared/bots.ts | 91 ++++ src/shared/ipc.ts | 20 +- src/shared/telemetry.ts | 17 +- src/shared/tool-history.ts | 6 +- src/shared/tools.ts | 56 +- src/shared/types.ts | 36 +- test/bots-shared.ts | 44 ++ test/bots-ui.cjs | 229 ++++++++ test/bots.ts | 420 +++++++++++++++ test/canvas/BotsHarness.tsx | 210 ++++++++ test/canvas/README.md | 1 + test/canvas/bots-smoke.cjs | 138 +++++ test/canvas/bridge.ts | 8 + test/canvas/diff.ts | 38 ++ test/canvas/harness.tsx | 15 +- test/shared.ts | 22 +- test/smoke.ts | 55 +- 61 files changed, 4559 insertions(+), 1392 deletions(-) create mode 100644 BOTS.md create mode 100644 src/main/db/bots.ts create mode 100644 src/main/harness/bot-tools.ts create mode 100644 src/main/services/automation.ts delete mode 100644 src/main/services/loops.ts create mode 100644 src/main/services/turn-state.ts create mode 100644 src/renderer/src/components/BotAvatar.tsx create mode 100644 src/renderer/src/components/BotSettingsPane.tsx create mode 100644 src/renderer/src/components/BotsSection.tsx delete mode 100644 src/renderer/src/components/LoopDetailsPane.tsx delete mode 100644 src/renderer/src/components/LoopsSection.tsx create mode 100644 src/shared/bots.ts create mode 100644 test/bots-shared.ts create mode 100644 test/bots-ui.cjs create mode 100644 test/bots.ts create mode 100644 test/canvas/BotsHarness.tsx create mode 100644 test/canvas/bots-smoke.cjs diff --git a/BOTS.md b/BOTS.md new file mode 100644 index 0000000..f731e17 --- /dev/null +++ b/BOTS.md @@ -0,0 +1,68 @@ +# Bots + +Bots are persistent, top-level chats, not project sessions or temporary subagents. +Create one below **New project**, pick a unique username, and describe its role in +chat. The bot can save that role with `bot_manage`. Its model, conversation, +tools, services, browser, and queue use the existing session harness. + +## Collaboration + +- A leading `@username` in a session sends that request and a bounded slice of + session context to the bot's own chat. Its attributed answer appears back in + the source session. Mid-sentence mentions do not redirect a user's request. +- Agents use `bot_invoke` to delegate explicitly. The response is persisted in + the caller's transcript and a continuation is queued for the caller. +- `project_list` and `session_manage` discover projects, create/read/update/delete + project sessions, and send prompts to them. New sessions honor workstream + isolation. Busy sessions cannot be deleted through these tools. +- `queue_manage` creates, lists, reads, edits/retries, and deletes pending + messages in any session or bot chat. It supports delayed delivery. +- Handoffs and continuations carry a maximum eight-hop budget. A result does + not recursively reply to its sender. Each target has a maximum 100 queued + messages and at most one active turn; automation runs at most four targets + concurrently. + +## Scheduling + +`bot_schedule` and bot settings support multiple jobs per bot: + +- Intervals of at least one minute. +- Five-field cron expressions with an explicit IANA timezone. +- Explicit epoch-millisecond timestamps. +- Optional remaining-run limits and pause/resume. + +The main process atomically enqueues each due prompt and advances its schedule. +No renderer needs to be open. **Roxy must still be running; this is not an OS or +cloud scheduler.** Missed interval/cron beats coalesce into one delivery after +startup. Explicit timestamps remain distinct deliveries. +Pausing affects future beats; already queued messages remain editable in the +queue. Deleting a schedule cancels its not-yet-started pending deliveries. + +The main process owns queue consumption for desktop, phone, bots, and scheduled +jobs. Failed requests remain in the queue with an error and block later work +until edited/retried or removed. Stop pauses draining. Interrupted deliveries +are marked failed on startup rather than replaying potentially non-idempotent +tool actions. There is no exactly-once guarantee for external side effects. + +New bots get private working folders under the app's user-data directory, with +the full existing harness and its normal tool restrictions. They coordinate +project work through sessions rather than silently borrowing the currently open +project. Migrated loops retain their working directory so existing tasks keep +working, but appear in the top-level bot navigation. + +## Migration And Checks + +Schema v24 migrates existing loops to bots and interval jobs, retaining chat IDs, +transcripts, inference settings, workspace paths, pending messages, enabled +state, and schedule times. Colliding legacy names get unique usernames. The old +loop scheduler, tools, IPC, and navigation are removed. + +The useful ideas from PR #91 are preserved: persistent identities, standing +roles, leading-mention routing, and attributed handoffs. Channel membership and +prose-scanning automatic handoffs are deliberately not required. Bots always +run in their own conversation; explicit tools handle agent-to-agent delegation. + +Run `npm run smoke:bots` for scheduling/mention unit checks and isolated Electron +runtime tests covering migration, CRUD, queue ownership, failures, handoffs, and +the real harness with a deterministic model transport. No live model credentials +are used by these tests. diff --git a/README.md b/README.md index 43e0a15..1e6abfa 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ roxy/ │ ├── main/ # Electron main process (Node.js) │ │ ├── index.ts # App lifecycle, window creation, service startup │ │ ├── harness/ # The agent loop: agent.ts (loop + tool schemas), tools.ts (dispatch) -│ │ ├── services/ # llm.ts, aisdk.ts, mcp.ts, lsp.ts, skills.ts, browser.ts, loops.ts, … +│ │ ├── services/ # llm.ts, aisdk.ts, mcp.ts, lsp.ts, skills.ts, browser.ts, automation.ts, … │ │ ├── db/ # better-sqlite3 store: schema, migrations, repo │ │ └── ipc/ # ipcMain handlers wiring the renderer to the harness/services │ ├── preload/ # Secure bridge between main and renderer (window.api) @@ -108,8 +108,8 @@ The main process runs a single provider-agnostic agent loop; the renderer only s ([`services/mcp.ts`](src/main/services/mcp.ts)), language-server diagnostics fed back after edits ([`services/lsp.ts`](src/main/services/lsp.ts)), and on-demand `SKILL.md` skills ([`services/skills.ts`](src/main/services/skills.ts)). Roxy's own differentiators — the persistent - browser toolset ([`services/browser.ts`](src/main/services/browser.ts)) and recurring "loops" - ([`services/loops.ts`](src/main/services/loops.ts)) — run through the same loop. + browser toolset ([`services/browser.ts`](src/main/services/browser.ts)) and persistent [bots](BOTS.md) + ([`services/automation.ts`](src/main/services/automation.ts)) run through the same harness. ### Remote Workspace diff --git a/package-lock.json b/package-lock.json index c7c2edf..2fc52eb 100644 --- a/package-lock.json +++ b/package-lock.json @@ -19,10 +19,12 @@ "ai": "^5.0.210", "better-sqlite3": "^12.11.1", "clsx": "^2.1.1", + "cron-parser": "^5.10.0", "d3-scale": "^4.0.2", "d3-shape": "^3.2.0", "diff": "^9.0.0", "electron-updater": "^6.3.9", + "facehash": "^0.1.0", "i18next": "^25.10.10", "lucide-react": "^1.21.0", "morphicons": "^1.7.0", @@ -4722,6 +4724,18 @@ "node": ">= 10" } }, + "node_modules/cron-parser": { + "version": "5.10.0", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.10.0.tgz", + "integrity": "sha512-izNAxJyRWUP8ljBoDSub5WyrVOUlT4SLGShswE7eoRBpp6QUsSycYxLBMJlbshgPBMcPT/nrfgjNY2918ayv2A==", + "license": "MIT", + "dependencies": { + "luxon": "^3.7.2" + }, + "engines": { + "node": ">=18" + } + }, "node_modules/cross-spawn": { "version": "7.0.6", "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", @@ -5899,6 +5913,26 @@ "license": "MIT", "optional": true }, + "node_modules/facehash": { + "version": "0.1.0", + "resolved": "https://registry.npmjs.org/facehash/-/facehash-0.1.0.tgz", + "integrity": "sha512-tv/QVZjLvEXHssqBaJECq+kRLFwwhd017PKk8ucT7aLingL2OZ5zEqKwPMHmT9+YQO92MVFWGZQP6vxV+P5vrQ==", + "license": "MIT", + "peerDependencies": { + "@types/react": "", + "next": ">=15", + "react": ">=18 <20", + "react-dom": ">=18 <20" + }, + "peerDependenciesMeta": { + "@types/react": { + "optional": true + }, + "next": { + "optional": true + } + } + }, "node_modules/fast-deep-equal": { "version": "3.1.3", "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz", @@ -7474,6 +7508,15 @@ "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, + "node_modules/luxon": { + "version": "3.7.2", + "resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz", + "integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==", + "license": "MIT", + "engines": { + "node": ">=12" + } + }, "node_modules/magic-string": { "version": "0.30.21", "resolved": "https://registry.npmjs.org/magic-string/-/magic-string-0.30.21.tgz", diff --git a/package.json b/package.json index d6cfb9a..5a524b6 100644 --- a/package.json +++ b/package.json @@ -45,6 +45,7 @@ "smoke:cliproxy": "esbuild test/cliproxy.ts --bundle --platform=node --format=cjs --packages=external --outfile=test/.out/cliproxy.cjs && electron test/.out/cliproxy.cjs", "worktree:setup": "npm ci --prefer-offline --no-audit --no-fund && electron-builder install-app-deps", "smoke:store": "node test/store-guard.mjs", + "smoke:bots": "esbuild test/bots-shared.ts --bundle --platform=node --format=cjs --packages=external --outfile=test/.out/bots-shared.cjs && node test/.out/bots-shared.cjs && esbuild src/preload/index.ts --bundle --platform=node --format=cjs --packages=external --outfile=test/.out/bots-preload.cjs && esbuild test/bots.ts --bundle --platform=node --format=cjs --packages=external --outfile=test/.out/bots.cjs && electron test/.out/bots.cjs", "smoke:cookies": "esbuild test/cookies.ts --bundle --platform=node --format=cjs --packages=external --outfile=test/.out/cookies.cjs && electron test/.out/cookies.cjs", "smoke:i18n": "esbuild test/i18n.ts --bundle --platform=node --format=cjs --outfile=test/.out/i18n.cjs && node test/.out/i18n.cjs", "i18n:translate": "node script/i18n-translate.mjs", @@ -64,10 +65,12 @@ "ai": "^5.0.210", "better-sqlite3": "^12.11.1", "clsx": "^2.1.1", + "cron-parser": "^5.10.0", "d3-scale": "^4.0.2", "d3-shape": "^3.2.0", "diff": "^9.0.0", "electron-updater": "^6.3.9", + "facehash": "^0.1.0", "i18next": "^25.10.10", "lucide-react": "^1.21.0", "morphicons": "^1.7.0", diff --git a/src/main/db/bots.ts b/src/main/db/bots.ts new file mode 100644 index 0000000..327c23a --- /dev/null +++ b/src/main/db/bots.ts @@ -0,0 +1,177 @@ +import { randomUUID } from 'node:crypto' +import { botUsername, nextBotRun, type Bot, type BotJob, type BotJobInput } from '../../shared/bots' +import { getDb } from './database' +import * as repo from './repo' + +const BOT_COLUMNS = 'id, username, instructions, chat_id AS chatId, created_at AS createdAt' +const JOB_COLUMNS = `id, bot_id AS botId, name, prompt, schedule, enabled, + next_run_at AS nextRunAt, last_run_at AS lastRunAt, remaining_runs AS remainingRuns, created_at AS createdAt` + +export function listBots(): Bot[] { + return getDb().prepare(`SELECT ${BOT_COLUMNS} FROM bots ORDER BY created_at, id`).all() as Bot[] +} + +export function getBot(id: string): Bot | undefined { + return getDb() + .prepare(`SELECT ${BOT_COLUMNS} FROM bots WHERE id = ? OR username = ?`) + .get(id, id.replace(/^@/, '')) as Bot | undefined +} + +export function chatBot(chatId: string): Bot | undefined { + return getDb().prepare(`SELECT ${BOT_COLUMNS} FROM bots WHERE chat_id = ?`).get(chatId) as + | Bot + | undefined +} + +export function createBot(username: string, instructions = ''): Bot { + if (typeof username !== 'string' || typeof instructions !== 'string') + throw new Error('Bot username and instructions must be text') + username = botUsername(username) + if (getBot(username)) throw new Error('That bot username is already taken') + return getDb().transaction(() => { + const chat = repo.createChat({ title: username, kind: 'bot' }) + const bot: Bot = { + id: randomUUID(), + username, + instructions, + chatId: chat.id, + createdAt: Date.now() + } + getDb() + .prepare( + 'INSERT INTO bots(id, username, instructions, chat_id, created_at) VALUES (?, ?, ?, ?, ?)' + ) + .run(bot.id, bot.username, bot.instructions, bot.chatId, bot.createdAt) + return bot + })() +} + +export function updateBot(id: string, patch: { username?: string; instructions?: string }): Bot { + const bot = getBot(id) + if (!bot) throw new Error('Bot not found') + const username = patch.username === undefined ? bot.username : botUsername(patch.username) + const other = getBot(username) + if (other && other.id !== bot.id) throw new Error('That bot username is already taken') + if (patch.instructions !== undefined && typeof patch.instructions !== 'string') + throw new Error('Instructions must be text') + return getDb().transaction(() => { + getDb() + .prepare('UPDATE bots SET username = ?, instructions = ? WHERE id = ?') + .run(username, patch.instructions ?? bot.instructions, bot.id) + if (username !== bot.username) repo.renameChat(bot.chatId, username) + return getBot(bot.id)! + })() +} + +export function removeBot(id: string): void { + const bot = getBot(id) + if (!bot) throw new Error('Bot not found') + repo.removeChat(bot.chatId) +} + +export function listJobs(botId?: string): BotJob[] { + const rows = ( + botId + ? getDb() + .prepare(`SELECT ${JOB_COLUMNS} FROM bot_jobs WHERE bot_id = ? ORDER BY created_at, id`) + .all(botId) + : getDb().prepare(`SELECT ${JOB_COLUMNS} FROM bot_jobs ORDER BY created_at, id`).all() + ) as (Omit & { schedule: string; enabled: number })[] + return rows.map((row) => ({ ...row, schedule: JSON.parse(row.schedule), enabled: !!row.enabled })) +} + +export function saveJob(input: BotJobInput, id?: string): BotJob { + const bot = getBot(input.botId) + if (!bot) throw new Error('Bot not found') + const old = id ? listJobs(bot.id).find((job) => job.id === id) : undefined + if (id && !old) throw new Error('Schedule not found for this bot') + if (!input.name?.trim() || !input.prompt?.trim()) + throw new Error('A schedule needs a name and prompt') + const remainingRuns = + input.remainingRuns === undefined ? (old?.remainingRuns ?? null) : input.remainingRuns + if ( + remainingRuns !== null && + (!Number.isSafeInteger(remainingRuns) || + remainingRuns < 0 || + (remainingRuns === 0 && input.enabled !== false)) + ) { + throw new Error('Run count must be a positive whole number, or null for unlimited') + } + const now = Date.now() + const enabled = input.enabled ?? old?.enabled ?? true + const computedNext = nextBotRun(input.schedule, now) + const next = + old?.enabled && enabled && JSON.stringify(old.schedule) === JSON.stringify(input.schedule) + ? old.nextRunAt + : computedNext + if (enabled && next === null) throw new Error('This schedule has no future runs') + const job: BotJob = { + id: id ?? randomUUID(), + botId: bot.id, + name: input.name.trim(), + prompt: input.prompt, + schedule: input.schedule, + enabled, + nextRunAt: next, + lastRunAt: old?.lastRunAt ?? null, + remainingRuns, + createdAt: old?.createdAt ?? now + } + getDb() + .prepare( + `INSERT INTO bot_jobs(id, bot_id, name, prompt, schedule, enabled, next_run_at, last_run_at, remaining_runs, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET name = excluded.name, prompt = excluded.prompt, schedule = excluded.schedule, + enabled = excluded.enabled, next_run_at = excluded.next_run_at, remaining_runs = excluded.remaining_runs` + ) + .run( + job.id, + job.botId, + job.name, + job.prompt, + JSON.stringify(job.schedule), + Number(job.enabled), + job.nextRunAt, + job.lastRunAt, + job.remainingRuns, + job.createdAt + ) + return job +} + +export function removeJob(id: string): void { + getDb().transaction(() => { + getDb().prepare(`DELETE FROM queue WHERE schedule_id = ? AND state = 'pending'`).run(id) + getDb().prepare('DELETE FROM bot_jobs WHERE id = ?').run(id) + })() +} + +/** Advance and enqueue in one transaction: a crash cannot consume a beat without a delivery. */ +export function enqueueDueJobs(now = Date.now()): string[] { + return getDb().transaction(() => { + const chats = new Set() + for (const job of listJobs()) { + if (!job.enabled || job.nextRunAt === null || job.nextRunAt > now) continue + const bot = getBot(job.botId) + if (!bot) continue + const queued = getDb() + .prepare('SELECT COUNT(*) AS n FROM queue WHERE chat_id = ?') + .get(bot.chatId) as { n: number } + if (queued.n >= 100) continue + const next = nextBotRun( + job.schedule, + job.schedule.kind === 'timestamps' ? job.nextRunAt : now + ) + const remaining = job.remainingRuns === null ? null : job.remainingRuns - 1 + const item = repo.enqueue(bot.chatId, job.prompt) + getDb().prepare('UPDATE queue SET schedule_id = ? WHERE id = ?').run(job.id, item.id) + getDb() + .prepare( + 'UPDATE bot_jobs SET last_run_at = ?, next_run_at = ?, remaining_runs = ?, enabled = ? WHERE id = ?' + ) + .run(now, next, remaining, Number(next !== null && remaining !== 0), job.id) + chats.add(bot.chatId) + } + return [...chats] + })() +} diff --git a/src/main/db/migrations.ts b/src/main/db/migrations.ts index 401482d..b169da4 100644 --- a/src/main/db/migrations.ts +++ b/src/main/db/migrations.ts @@ -6,6 +6,42 @@ import type { Database } from 'better-sqlite3' */ export type Migration = string | ((db: Database) => void) +function botSchema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS bots ( + id TEXT PRIMARY KEY, + username TEXT NOT NULL COLLATE NOCASE UNIQUE, + instructions TEXT NOT NULL DEFAULT '', + chat_id TEXT NOT NULL UNIQUE REFERENCES chats(id) ON DELETE CASCADE, + created_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS bot_jobs ( + id TEXT PRIMARY KEY, + bot_id TEXT NOT NULL REFERENCES bots(id) ON DELETE CASCADE, + name TEXT NOT NULL, + prompt TEXT NOT NULL, + schedule TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1, + next_run_at INTEGER, + last_run_at INTEGER, + remaining_runs INTEGER, + created_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_bot_jobs_due ON bot_jobs(enabled, next_run_at); + `) + addColumnIfMissing(db, 'messages', 'bot_id', 'TEXT') + addColumnIfMissing(db, 'messages', 'bot_username', 'TEXT') + addColumnIfMissing(db, 'queue', 'source_chat_id', 'TEXT') + addColumnIfMissing(db, 'queue', 'reply_to_chat_id', 'TEXT') + addColumnIfMissing(db, 'queue', 'hops', 'INTEGER NOT NULL DEFAULT 0') + addColumnIfMissing(db, 'queue', 'not_before', 'INTEGER NOT NULL DEFAULT 0') + addColumnIfMissing(db, 'queue', 'state', `TEXT NOT NULL DEFAULT 'pending'`) + addColumnIfMissing(db, 'queue', 'error', 'TEXT') + addColumnIfMissing(db, 'queue', 'message_id', 'TEXT') + addColumnIfMissing(db, 'queue', 'continue_reply', 'INTEGER NOT NULL DEFAULT 0') + addColumnIfMissing(db, 'queue', 'schedule_id', 'TEXT') +} + /** Whether a table already has a column — SQLite can't express this in DDL. */ export function hasColumn(db: Database, table: string, column: string): boolean { const cols = db.prepare(`PRAGMA table_info(${table})`).all() as { name: string }[] @@ -493,7 +529,68 @@ export const MIGRATIONS: Migration[] = [ hidden_at INTEGER NOT NULL, PRIMARY KEY (provider_id, model) ); - ` + `, + + // ---- v24: global bots and main-process scheduled delivery ---- + (db) => { + botSchema(db) + const loops = db.prepare('SELECT * FROM loops ORDER BY created_at, id').all() as { + id: string + name: string + prompt: string + chat_id: string + interval_minutes: number + enabled: number + next_run_at: number + last_run_at: number | null + created_at: number + }[] + const taken = new Set( + (db.prepare('SELECT username FROM bots').all() as { username: string }[]).map( + (b) => b.username + ) + ) + for (const loop of loops) { + let base = loop.name + .toLowerCase() + .replace(/[^a-z0-9_-]+/g, '-') + .replace(/^-+|-+$/g, '') + .slice(0, 24) + if (!/^[a-z]/.test(base) || base.length < 2 || base === 'roxy') base = `bot-${base || 'loop'}` + let username = base + for (let n = 2; taken.has(username); n++) username = `${base}-${n}` + taken.add(username) + db.prepare( + 'INSERT INTO bots(id, username, instructions, chat_id, created_at) VALUES (?, ?, ?, ?, ?)' + ).run( + loop.id, + username, + `Migrated from the loop ${loop.name}. Your scheduled task is configured separately.`, + loop.chat_id, + loop.created_at + ) + db.prepare('UPDATE chats SET kind = ? WHERE id = ?').run('bot', loop.chat_id) + db.prepare( + `INSERT INTO bot_jobs(id, bot_id, name, prompt, schedule, enabled, next_run_at, last_run_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)` + ).run( + loop.id, + loop.id, + loop.name, + loop.prompt, + JSON.stringify({ + kind: 'interval', + minutes: Math.max(1, Math.min(525600, loop.interval_minutes || 1)) + }), + loop.enabled, + loop.next_run_at, + loop.last_run_at, + loop.created_at + ) + } + // Keep transcripts and queues in place; only the obsolete scheduler rows go away. + db.exec(`UPDATE chats SET kind = 'main' WHERE kind = 'loop'; DELETE FROM loops;`) + } ] /** @@ -519,6 +616,7 @@ export const MIGRATIONS: Migration[] = [ */ export function repairSchema(db: Database): void { db.exec(REPAIR_SCHEMA_SQL) + botSchema(db) // Columns added by later migrations: CREATE TABLE IF NOT EXISTS won't add // them to a table that already exists. addColumnIfMissing(db, 'chats', 'worktree_path', 'TEXT') diff --git a/src/main/db/repo.ts b/src/main/db/repo.ts index 0b0f9be..912ad5f 100644 --- a/src/main/db/repo.ts +++ b/src/main/db/repo.ts @@ -11,7 +11,6 @@ import type { ConnectedProvider, ConnectProviderInput, IntegrationConnection, - Loop, Message, MessagePart, MessageRole, @@ -28,7 +27,7 @@ import type { WorktreeIntent } from '../../shared/types' import { parseRepoLinks, serializeRepoLinks, type RepoLink } from '../../shared/repos' -import type { CreateChatInput, CreateLoopInput } from '../../shared/api' +import type { CreateChatInput } from '../../shared/api' import { parseReasoningEffort, seedSessionConfig, @@ -85,6 +84,8 @@ interface MessageRow { content: string parts: string | null created_at: number + bot_id: string | null + bot_username: string | null } interface IntegrationRow { @@ -802,7 +803,7 @@ export function createChat(input: CreateChatInput = {}): Chat { input.kind ?? 'main', providerId, model, - seed.agentId, + input.kind === 'bot' ? 'build' : seed.agentId, seed.reasoningEffort, seed.contextLimit, input.workspacePath ?? null, @@ -815,7 +816,7 @@ export function createChat(input: CreateChatInput = {}): Chat { // A new main session or loop in a workspace registers that project (appended // to the bottom of the project list) the first time we see that folder. Sub- // agent sessions group under their parent, so they never register a project. - if (input.workspacePath && (input.kind ?? 'main') !== 'sub') ensureProject(input.workspacePath) + if (input.workspacePath && (input.kind ?? 'main') === 'main') ensureProject(input.workspacePath) const chat = getChat(id) if (!chat) throw new Error('Failed to create chat') return chat @@ -870,11 +871,11 @@ export function forkChat(sourceId: string, input: { title?: string } = {}): Chat const now = Date.now() const title = input.title?.trim() || `${source.title} (fork)` const messages = db - .prepare('SELECT role, content, parts, created_at FROM messages WHERE chat_id = ?') - .all(sourceId) as Pick[] + .prepare('SELECT * FROM messages WHERE chat_id = ? ORDER BY created_at, rowid') + .all(sourceId) as MessageRow[] const insertMessage = db.prepare( - 'INSERT INTO messages(id, chat_id, role, content, parts, created_at) VALUES(?, ?, ?, ?, ?, ?)' + 'INSERT INTO messages(id, chat_id, role, content, parts, created_at, bot_id, bot_username) VALUES(?, ?, ?, ?, ?, ?, ?, ?)' ) db.transaction(() => { db.prepare( @@ -897,7 +898,16 @@ export function forkChat(sourceId: string, input: { title?: string } = {}): Chat now ) for (const m of messages) { - insertMessage.run(randomUUID(), id, m.role, m.content, m.parts, m.created_at) + insertMessage.run( + randomUUID(), + id, + m.role, + m.content, + m.parts, + m.created_at, + m.bot_id, + m.bot_username + ) } })() @@ -1110,9 +1120,7 @@ export function ensureProject(path: string): void { export function pruneProjectIfEmpty(path: string): void { const db = getDb() const { n } = db - .prepare( - "SELECT COUNT(*) AS n FROM chats WHERE workspace_path IS ? AND kind IN ('main', 'loop')" - ) + .prepare("SELECT COUNT(*) AS n FROM chats WHERE workspace_path IS ? AND kind = 'main'") .get(path) as { n: number } if (n === 0) db.prepare('DELETE FROM projects WHERE path = ?').run(path) } @@ -1168,13 +1176,15 @@ function rowToMessage(row: MessageRow): Message { role: row.role as MessageRole, content: row.content, parts: parseParts(row.parts, row.content), - createdAt: row.created_at + createdAt: row.created_at, + ...(row.bot_id ? { botId: row.bot_id } : {}), + ...(row.bot_username ? { botUsername: row.bot_username } : {}) } } export function listMessages(chatId: string): Message[] { const rows = getDb() - .prepare('SELECT * FROM messages WHERE chat_id = ? ORDER BY created_at ASC') + .prepare('SELECT * FROM messages WHERE chat_id = ? ORDER BY created_at ASC, rowid ASC') .all(chatId) as MessageRow[] return rows.map(rowToMessage) } @@ -1187,8 +1197,17 @@ export function addMessage(input: AddMessageInput): Message { const db = getDb() const tx = db.transaction(() => { db.prepare( - 'INSERT INTO messages(id, chat_id, role, content, parts, created_at) VALUES(?, ?, ?, ?, ?, ?)' - ).run(id, input.chatId, input.role, input.content, partsJson, now) + 'INSERT INTO messages(id, chat_id, role, content, parts, created_at, bot_id, bot_username) VALUES(?, ?, ?, ?, ?, ?, ?, ?)' + ).run( + id, + input.chatId, + input.role, + input.content, + partsJson, + now, + input.botId ?? null, + input.botUsername ?? null + ) db.prepare('UPDATE chats SET updated_at = ? WHERE id = ?').run(now, input.chatId) // One assistant message = one agent turn. Credited to the durable ledger in // the SAME transaction as the message, so the graph can never disagree with @@ -1204,114 +1223,12 @@ export function addMessage(input: AddMessageInput): Message { role: input.role, content: input.content, parts, - createdAt: now - } -} - -// ---- Loops ------------------------------------------------------------------- - -interface LoopRow { - id: string - name: string - prompt: string - interval_minutes: number - enabled: number - chat_id: string - last_run_at: number | null - next_run_at: number - created_at: number -} - -function rowToLoop(row: LoopRow): Loop { - return { - id: row.id, - name: row.name, - prompt: row.prompt, - intervalMinutes: row.interval_minutes, - enabled: row.enabled > 0, - chatId: row.chat_id, - lastRunAt: row.last_run_at, - nextRunAt: row.next_run_at, - createdAt: row.created_at + createdAt: now, + ...(input.botId ? { botId: input.botId } : {}), + ...(input.botUsername ? { botUsername: input.botUsername } : {}) } } -function getLoop(id: string): Loop | undefined { - const row = getDb().prepare('SELECT * FROM loops WHERE id = ?').get(id) as LoopRow | undefined - return row ? rowToLoop(row) : undefined -} - -export function listLoops(): Loop[] { - const rows = getDb().prepare('SELECT * FROM loops ORDER BY created_at DESC').all() as LoopRow[] - return rows.map(rowToLoop) -} - -export function createLoop(input: CreateLoopInput): Loop { - const id = randomUUID() - const now = Date.now() - const interval = Math.max(1, Math.floor(input.intervalMinutes)) - const name = input.name.trim() || 'Loop' - const chat = createChat({ title: name, kind: 'loop', workspacePath: input.workspacePath ?? null }) - getDb() - .prepare( - `INSERT INTO loops(id, name, prompt, interval_minutes, enabled, chat_id, last_run_at, next_run_at, created_at) - VALUES(?, ?, ?, ?, 1, ?, NULL, ?, ?)` - ) - .run(id, name, input.prompt, interval, chat.id, now, now) - const loop = getLoop(id) - if (!loop) throw new Error('Failed to create loop') - return loop -} - -export function setLoopEnabled(id: string, enabled: boolean): void { - if (enabled) { - getDb() - .prepare('UPDATE loops SET enabled = 1, next_run_at = ? WHERE id = ?') - .run(Date.now(), id) - } else { - getDb().prepare('UPDATE loops SET enabled = 0 WHERE id = ?').run(id) - } -} - -export function removeLoop(id: string): void { - const loop = getLoop(id) - if (!loop) return - // The PROJECT folder (not sessionCwd) — same reason as removeChat. - const workspace = getChatWorkspace(loop.chatId) - // Deleting the chat cascades to the loop row and its messages. - getDb().prepare('DELETE FROM chats WHERE id = ?').run(loop.chatId) - if (workspace) pruneProjectIfEmpty(workspace) -} - -export function dueLoops(now: number): Loop[] { - const rows = getDb() - .prepare('SELECT * FROM loops WHERE enabled = 1 AND next_run_at <= ? ORDER BY next_run_at ASC') - .all(now) as LoopRow[] - return rows.map(rowToLoop) -} - -/** Append one heartbeat run (scheduled prompt + response) and schedule the next. */ -export function appendLoopRun(loopId: string, userContent: string, assistantContent: string): void { - const loop = getLoop(loopId) - if (!loop) return - const now = Date.now() - addMessage({ chatId: loop.chatId, role: 'user', content: userContent }) - addMessage({ chatId: loop.chatId, role: 'assistant', content: assistantContent }) - getDb() - .prepare('UPDATE loops SET last_run_at = ?, next_run_at = ? WHERE id = ?') - .run(now, now + loop.intervalMinutes * 60_000, loopId) -} - -/** Advance a loop's schedule after a beat fires (the agent turn runs separately). */ -export function markLoopRan(loopId: string): void { - const loop = getLoop(loopId) - if (!loop) return - const now = Date.now() - getDb() - .prepare('UPDATE loops SET last_run_at = ?, next_run_at = ? WHERE id = ?') - .run(now, now + loop.intervalMinutes * 60_000, loopId) -} - // ---- Sessions status (list_sessions / check_session tools) ------------------- export function listSessionsStatus(): SessionStatus[] { @@ -1355,18 +1272,30 @@ interface QueueRow { content: string images: string | null created_at: number + source_chat_id: string | null + reply_to_chat_id: string | null + hops: number + not_before: number + state: 'pending' | 'running' | 'failed' + error: string | null } export function listQueue(chatId: string): QueueItem[] { const rows = getDb() - .prepare('SELECT * FROM queue WHERE chat_id = ? ORDER BY created_at ASC') + .prepare('SELECT * FROM queue WHERE chat_id = ? ORDER BY created_at ASC, rowid ASC') .all(chatId) as QueueRow[] return rows.map((r) => ({ id: r.id, chatId: r.chat_id, content: r.content, ...(r.images ? { images: JSON.parse(r.images) as QueueImage[] } : {}), - createdAt: r.created_at + createdAt: r.created_at, + sourceChatId: r.source_chat_id ?? undefined, + replyToChatId: r.reply_to_chat_id ?? undefined, + hops: r.hops, + notBefore: r.not_before, + state: r.state, + error: r.error ?? undefined })) } @@ -1381,7 +1310,12 @@ export function enqueue(chatId: string, content: string, images?: QueueImage[]): } export function removeQueueItem(id: string): void { - getDb().prepare('DELETE FROM queue WHERE id = ?').run(id) + const row = getDb().prepare('SELECT state FROM queue WHERE id = ?').get(id) as + | { state: string } + | undefined + if (row?.state === 'running') + throw new Error('Stop the session before removing its running message') + getDb().prepare(`DELETE FROM queue WHERE id = ? AND state != 'running'`).run(id) } /** Edit a queued item's text + images in place, keeping its `created_at` (so its @@ -1391,19 +1325,31 @@ export function updateQueueItem( content: string, images?: QueueImage[] ): QueueItem | undefined { + if (!content.trim() && !images?.length) throw new Error('A prompt is required') const imagesJson = images && images.length ? JSON.stringify(images) : null + const previous = getDb() + .prepare('SELECT content, images, message_id, state FROM queue WHERE id = ?') + .get(id) as + | { + content: string + images: string | null + message_id: string | null + state: string + } + | undefined + if (previous?.state === 'running') throw new Error('This message is already running') + // Retrying an unchanged request reuses its user bubble. Editing its content + // creates a new user turn, so the model receives the correction, not stale text. + const changed = previous && (previous.content !== content || previous.images !== imagesJson) getDb() - .prepare('UPDATE queue SET content = ?, images = ? WHERE id = ?') - .run(content, imagesJson, id) + .prepare( + `UPDATE queue SET content = ?, images = ?, state = 'pending', error = NULL, + message_id = CASE WHEN ? THEN NULL ELSE message_id END WHERE id = ? AND state != 'running'` + ) + .run(content, imagesJson, Number(!!changed), id) const row = getDb().prepare('SELECT * FROM queue WHERE id = ?').get(id) as QueueRow | undefined if (!row) return undefined - return { - id: row.id, - chatId: row.chat_id, - content: row.content, - ...(row.images ? { images: JSON.parse(row.images) as QueueImage[] } : {}), - createdAt: row.created_at - } + return listQueue(row.chat_id).find((item) => item.id === id) } /** Reorder a chat's queue to match `orderedIds` (front = runs next). Assigns @@ -1412,13 +1358,14 @@ export function updateQueueItem( * chat's queue ids is passed. */ export function reorderQueue(chatId: string, orderedIds: string[]): void { const db = getDb() + if (db.prepare(`SELECT 1 FROM queue WHERE chat_id = ? AND state = 'running'`).get(chatId)) return const existing = db.prepare('SELECT id FROM queue WHERE chat_id = ?').all(chatId) as { id: string }[] if (existing.length < 2) return const valid = new Set(existing.map((r) => r.id)) const ids = orderedIds.filter((id) => valid.has(id)) - if (ids.length !== existing.length) return + if (ids.length !== existing.length || new Set(ids).size !== ids.length) return const update = db.prepare('UPDATE queue SET created_at = ? WHERE id = ?') db.transaction(() => ids.forEach((id, i) => update.run(i + 1, id)))() } diff --git a/src/main/harness/agent.ts b/src/main/harness/agent.ts index 5fb2083..1d3ace2 100644 --- a/src/main/harness/agent.ts +++ b/src/main/harness/agent.ts @@ -42,6 +42,7 @@ import { import { existsSync, readFileSync } from 'node:fs' import { dirname, join, resolve } from 'node:path' import * as repo from '../db/repo' +import { chatBot, listBots } from '../db/bots' import { runTool } from './tools' import { boundToolOutput } from '../services/tool-output-store' import { modelCost } from '../services/models' @@ -481,6 +482,29 @@ function buildSystemMessage( ...(mcpInfo ? [mcpInfo] : []), ...(agentPrompt ? [agentPrompt] : []) ] + const bot = chatId ? chatBot(chatId) : undefined + extra.push( + [ + '', + bot + ? `You are @${bot.username}, a persistent, top-level bot. Your session ID is ${chatId}.` + : `This session ID is ${chatId ?? 'unknown'}.`, + bot + ? `Your standing role (not an instruction to start working on every greeting):\n${bot.instructions || 'Ask the user what they want you to be or do. Use bot_manage to save the agreed role as instructions.'}` + : '', + 'Bots are local Roxy collaborators, not GitHub users or temporary task subagents.', + 'Use project_list and session_manage to discover projects and sessions. Use bot_manage to discover or configure persistent bots.', + 'Use bot_invoke to ask another bot for help. Its result is delivered to your transcript asynchronously; do not poll or duplicate its work.', + 'Use session_manage action send to prompt a project session. Use queue_manage for delayed messages, inspection, edits, cancellation, and retries.', + 'Use bot_schedule to configure optional interval, five-field cron (with timezone), or timestamp jobs. Never claim a schedule exists until the tool succeeds.', + 'A user message beginning with @username explicitly addresses that bot. Mentions inside prose are references, not handoffs. To hand off as an agent, call bot_invoke explicitly.', + 'Do not reflexively reply to a returned result by invoking its sender again. Keep collaboration finite. Ask before repeating a completed chain.', + ...listBots().map((b) => `@${b.username}: id=${b.id}, chat=${b.chatId}`), + '' + ] + .filter(Boolean) + .join('\n') + ) const contextSummary = chatId ? (repo.getChat(chatId)?.contextSummary ?? undefined) : undefined return assembleSystemPrompt({ base, @@ -639,24 +663,93 @@ const BASE_SCHEMAS = [ ['id'] ), fn('browser_close', 'Close the built-in browser and end the current browsing session.', {}, []), + fn('project_list', 'List all projects across Roxy with their workspace paths.', {}, []), fn( - 'loop_create', - 'Create a scheduled loop (a recurring "heartbeat") that re-runs a prompt in THIS project every N minutes — the agent runs fully each beat. Use when the user wants ongoing/recurring/autonomous/looping work (e.g. "every 5 min, keep improving the site").', + 'session_manage', + 'Create, list, read, update, delete, stop, or send a prompt to a session in any project. send queues work and returns its result to this chat asynchronously.', { - name: str('Short label for the loop.'), - prompt: str('The instruction to run every interval.'), - interval_minutes: { type: 'number', description: 'Minutes between runs (>= 1).' } + action: { + type: 'string', + enum: ['list', 'create', 'read', 'update', 'delete', 'send', 'stop'] + }, + id: str('Session ID for read/update/delete/send/stop.'), + project: str('Project path from project_list. Required to create; optional filter for list.'), + title: str('Session title.'), + description: str('Session description.'), + prompt: str('Prompt to send.') }, - ['name', 'prompt', 'interval_minutes'] + ['action'] + ), + fn( + 'bot_manage', + 'Manage persistent top-level bots. Instructions are a standing role, not an immediate task. Use read for its chat history.', + { + action: { type: 'string', enum: ['list', 'create', 'read', 'update', 'delete'] }, + id: str('Bot ID or exact username.'), + username: str( + 'Unique username: 2-32 lowercase letters, digits, underscore or hyphen; starts with a letter.' + ), + instructions: str('Persistent role and behavior for the bot.') + }, + ['action'] + ), + fn( + 'bot_invoke', + 'Ask a persistent bot to do work in its own chat using its full harness and memory. Queues safely if busy. Returns immediately; the final reply is delivered to this chat. Do not poll or duplicate it.', + { + bot: str('Bot ID or exact username from bot_manage list.'), + prompt: str('Self-contained task and relevant context for the bot.') + }, + ['bot', 'prompt'] + ), + fn( + 'bot_schedule', + 'Manage scheduled bot prompts. Jobs persist and run while Roxy is open, even with no chat window. Intervals and cron coalesce missed beats; explicit timestamps run once each. remaining_runs caps deliveries.', + { + action: { type: 'string', enum: ['list', 'create', 'update', 'delete'] }, + id: str('Schedule ID for update/delete; optional bot ID filter for list.'), + bot: str('Bot ID or username; defaults to the current bot.'), + name: str('Schedule label.'), + prompt: str('Prompt for each run.'), + enabled: { type: 'boolean' }, + remaining_runs: { + type: ['integer', 'null'], + minimum: 1, + description: 'Number of deliveries remaining; null for unlimited.' + }, + schedule: { + type: 'object', + properties: { + kind: { type: 'string', enum: ['interval', 'cron', 'timestamps'] }, + minutes: { type: 'number', minimum: 1 }, + expression: str('Five-field cron, e.g. 0 9 * * 1-5.'), + timezone: str('IANA timezone required for cron, e.g. America/New_York.'), + timestamps: { + type: 'array', + items: { type: 'integer' }, + description: 'Epoch milliseconds for explicit runs.' + } + }, + required: ['kind'] + } + }, + ['action'] + ), + fn( + 'queue_manage', + 'Create, list, read, edit/retry, or delete queued prompts on yourself, another bot chat, or any project session. Running items cannot be edited/deleted. Failed items block the queue until edited or removed.', + { + action: { type: 'string', enum: ['list', 'create', 'read', 'update', 'delete'] }, + id: str('Queue item ID.'), + session: str('Target chat ID; defaults to this session.'), + prompt: str('Prompt content.'), + not_before: { + type: 'integer', + description: 'Do not deliver before this epoch millisecond timestamp.' + } + }, + ['action'] ), - fn('loop_list', 'List the scheduled loops and whether each is running.', {}, []), - fn('loop_enable', 'Resume a paused loop by name or id.', { loop: str('Loop name or id.') }, [ - 'loop' - ]), - fn('loop_disable', 'Pause a running loop by name or id.', { loop: str('Loop name or id.') }, [ - 'loop' - ]), - fn('loop_remove', 'Delete a loop by name or id.', { loop: str('Loop name or id.') }, ['loop']), fn( 'change_session_metadata', "Organize THIS session: set its `title` (shown in the sidebar), a one-line `description` of what it's about, and/or a `tasks` checklist you maintain as you work. Send the FULL tasks array each time — it REPLACES the previous list. Use it to rename a vaguely-named session and to track multi-step work (mark a task in_progress when you start it, completed when done). If the session has a workstream, setting `title` also renames its git branch to match — but only while that branch is still the auto-generated one and has never been pushed. Pass `branch` to choose the branch name yourself.", @@ -1380,7 +1473,7 @@ async function runLoop(o: LoopOptions): Promise { if (tc.name.startsWith('mcp__')) trackFeature(metricsId, 'mcp_server') else if (tc.name === SKILL_TOOL_NAME) trackFeature(metricsId, 'skill') else if (tc.name.startsWith('browser_')) trackFeature(metricsId, 'browser') - else if (tc.name.startsWith('loop_')) trackFeature(metricsId, 'loop') + else if (tc.name.startsWith('bot_')) trackFeature(metricsId, 'bot') // Full output still streams to the UI (tool-end above); for the model's // rolling context, spill oversized results to disk and keep a head/tail // preview + a read-tool pointer instead of a blind 8k cut (Phase 9.3). diff --git a/src/main/harness/bot-tools.ts b/src/main/harness/bot-tools.ts new file mode 100644 index 0000000..0121067 --- /dev/null +++ b/src/main/harness/bot-tools.ts @@ -0,0 +1,229 @@ +import type { ToolResult } from '../../shared/types' +import type { BotJobInput } from '../../shared/bots' +import type { ToolContext } from './tools' +import * as bots from '../db/bots' +import * as repo from '../db/repo' +import { getDb } from '../db/database' +import { enqueuePrompt, notifyAutomation, notifyBots } from '../services/automation' +import { claimTurn, sessionBusy, stopTurn, resumeQueue } from '../services/turn-state' +import { emitSessionsUpdated } from '../services/session-events' +import { cancelSessionBackgroundJobs } from '../services/background-tasks' +import { endSubagentRuns } from '../services/subagent-stream' +import { killSessionBackground } from './tools' +import { disposeSession } from '../services/browser' +import { removeWorktreeForChat } from '../services/worktree' + +const text = (value: unknown): string => (typeof value === 'string' ? value : '') + +export async function runBotTool( + name: string, + input: Record, + ctx: ToolContext +): Promise { + const action = text(input.action) + const id = text(input.id) + const source = ctx.sessionId + const running = source + ? (getDb() + .prepare(`SELECT hops FROM queue WHERE chat_id = ? AND state = 'running'`) + .get(repo.rootSessionId(source)) as { hops: number } | undefined) + : undefined + const hops = (running?.hops ?? 0) + 1 + let result: unknown + switch (name) { + case 'project_list': + result = repo.listProjectOrder().map((path) => ({ + path, + sessions: repo.listChats().filter((c) => c.kind === 'main' && c.workspacePath === path) + .length + })) + break + case 'bot_manage': { + if (action === 'list') result = bots.listBots() + else if (action === 'read') { + const bot = bots.getBot(id) + if (!bot) throw new Error('Bot not found') + result = { + ...bot, + messages: repo.listMessages(bot.chatId).slice(-50), + jobs: bots.listJobs(bot.id) + } + } else if (action === 'create') + result = bots.createBot(text(input.username), text(input.instructions)) + else if (action === 'update') + result = bots.updateBot(id, { + ...(input.username !== undefined ? { username: text(input.username) } : {}), + ...(input.instructions !== undefined ? { instructions: text(input.instructions) } : {}) + }) + else if (action === 'delete') { + const bot = bots.getBot(id) + if (!bot) throw new Error('Bot not found') + if (bot.chatId === source || sessionBusy(bot.chatId)) + throw new Error('Stop the bot before deleting it; a bot cannot delete itself mid-turn') + cancelSessionBackgroundJobs(bot.chatId) + endSubagentRuns(bot.chatId) + killSessionBackground(bot.chatId) + disposeSession(bot.chatId) + bots.removeBot(id) + result = { deleted: id } + } else throw new Error('Unknown bot action') + notifyBots() + break + } + case 'bot_schedule': { + if (action === 'list') result = bots.listJobs(id ? (bots.getBot(id)?.id ?? id) : undefined) + else if (action === 'delete') { + const job = bots.listJobs().find((entry) => entry.id === id) + bots.removeJob(id) + const bot = job && bots.getBot(job.botId) + if (bot) notifyAutomation(bot.chatId) + result = { deleted: id } + } else if (action === 'create' || action === 'update') { + const old = action === 'update' ? bots.listJobs().find((job) => job.id === id) : undefined + if (action === 'update' && !old) throw new Error('Schedule not found') + const ref = text(input.bot) || old?.botId + const bot = ref ? bots.getBot(ref) : source ? bots.chatBot(source) : undefined + if (!bot) throw new Error('Name the bot to schedule') + result = bots.saveJob( + { + botId: bot.id, + name: text(input.name) || old?.name || '', + prompt: text(input.prompt) || old?.prompt || '', + schedule: (input.schedule ?? old?.schedule) as BotJobInput['schedule'], + enabled: input.enabled === undefined ? old?.enabled : !!input.enabled, + remainingRuns: + input.remaining_runs === undefined + ? old?.remainingRuns + : (input.remaining_runs as number | null) + }, + old?.id + ) + } else throw new Error('Unknown schedule action') + notifyBots() + break + } + case 'bot_invoke': { + const bot = bots.getBot(text(input.bot)) + if (!bot) throw new Error('Bot not found; use bot_manage list') + if (bot.chatId === source) throw new Error('Use queue_manage to queue work for yourself') + result = enqueuePrompt(bot.chatId, text(input.prompt), undefined, { + sourceChatId: source, + replyToChatId: source, + hops, + continueReply: true + }) + break + } + case 'session_manage': { + if (action === 'list') { + result = repo + .listChats() + .filter((c) => c.kind === 'main' && (!input.project || c.workspacePath === input.project)) + .map((c) => ({ ...c, running: sessionBusy(c.id) })) + } else if (action === 'create') { + const project = text(input.project) + if (!repo.listProjectOrder().includes(project)) + throw new Error('Choose an existing project from project_list') + result = repo.createChat({ + title: text(input.title) || 'New session', + workspacePath: project, + ...(repo.getSettings().autoWorkstream ? { worktree: { mode: 'new' as const } } : {}) + }) + } else { + const chat = repo.getChat(id) + if (!chat || chat.kind !== 'main') throw new Error('Project session not found') + if (action === 'read') + result = { + ...chat, + messages: repo.listMessages(id).slice(-100), + queue: repo.listQueue(id), + running: sessionBusy(id) + } + else if (action === 'update') { + repo.setChatMetadata(id, { + ...(input.title !== undefined ? { title: text(input.title) } : {}), + ...(input.description !== undefined ? { description: text(input.description) } : {}) + }) + result = repo.getChat(id) + } else if (action === 'delete') { + if (id === source || sessionBusy(id)) + throw new Error('Stop the session before deleting it') + cancelSessionBackgroundJobs(id) + endSubagentRuns(id) + killSessionBackground(id) + disposeSession(id) + const release = claimTurn(id, new AbortController())! + try { + await removeWorktreeForChat(id) + repo.removeChat(id) + } finally { + release() + } + result = { deleted: id } + } else if (action === 'send') { + result = enqueuePrompt(id, text(input.prompt), undefined, { + sourceChatId: source, + replyToChatId: source, + hops, + continueReply: id !== source + }) + } else if (action === 'stop') { + stopTurn(id) + result = { stopped: id } + } else throw new Error('Unknown session action') + } + emitSessionsUpdated({ reason: 'metadata', sessionIds: id ? [id] : [] }) + break + } + case 'queue_manage': { + const chatId = text(input.session) || source || '' + if (action === 'list') { + if (!repo.getChat(chatId)) throw new Error('Session not found') + result = repo.listQueue(chatId) + } else if (action === 'create') + result = enqueuePrompt(chatId, text(input.prompt), undefined, { + sourceChatId: source, + hops, + notBefore: input.not_before as number | undefined + }) + else { + const row = getDb().prepare('SELECT chat_id, state FROM queue WHERE id = ?').get(id) as + | { chat_id: string; state: string } + | undefined + if (!row) throw new Error('Queued message not found') + if (action === 'read') result = repo.listQueue(row.chat_id).find((item) => item.id === id) + else { + if (row.state === 'running') + throw new Error( + 'This message is running; stop its session before editing or deleting it' + ) + if (action === 'delete') { + repo.removeQueueItem(id) + result = { deleted: id } + } else if (action === 'update') { + const old = repo.listQueue(row.chat_id).find((item) => item.id === id)! + const prompt = input.prompt === undefined ? old.content : text(input.prompt) + if (!prompt.trim()) throw new Error('A prompt is required') + if ( + input.not_before !== undefined && + (!Number.isSafeInteger(input.not_before) || Number(input.not_before) < 0) + ) + throw new Error('Invalid not_before timestamp') + repo.updateQueueItem(id, prompt, old.images) + resumeQueue(row.chat_id) + if (input.not_before !== undefined) + getDb() + .prepare('UPDATE queue SET not_before = ? WHERE id = ?') + .run(input.not_before, id) + result = repo.listQueue(row.chat_id).find((item) => item.id === id) + } else throw new Error('Unknown queue action') + notifyAutomation(row.chat_id) + } + } + break + } + default: + throw new Error('Unknown bot tool') + } + return { ok: true, output: JSON.stringify(result, null, 2) } +} diff --git a/src/main/harness/tools.ts b/src/main/harness/tools.ts index 8f6bc74..39c2a7e 100644 --- a/src/main/harness/tools.ts +++ b/src/main/harness/tools.ts @@ -26,6 +26,7 @@ import { import * as browser from '../services/browser' import * as lsp from '../services/lsp' import * as repo from '../db/repo' +import { runBotTool } from './bot-tools' import { isManagedToolOutputPath } from '../services/tool-output-store' import { renderDiagnosticsBlock } from '../../shared/lsp' import { @@ -260,16 +261,13 @@ export async function runTool( case 'browser_close': browser.close(browserKey(ctx)) return { ok: true, output: 'Closed the browser.' } - case 'loop_create': - return runLoopCreate(input, ctx.cwd) - case 'loop_remove': - return runLoopRemove(str(input.loop ?? input.name ?? input.id)) - case 'loop_list': - return runLoopList() - case 'loop_enable': - return runLoopSet(str(input.loop ?? input.name ?? input.id), true) - case 'loop_disable': - return runLoopSet(str(input.loop ?? input.name ?? input.id), false) + case 'project_list': + case 'session_manage': + case 'bot_manage': + case 'bot_schedule': + case 'bot_invoke': + case 'queue_manage': + return await runBotTool(name, input, ctx) case 'change_session_metadata': return await runSetSessionMetadata(input, ctx.sessionId) case 'lsp': @@ -1201,69 +1199,6 @@ async function runBrowserType(selector: string, text: string, key?: string): Pro return { ok: !out.startsWith('No element') && !out.startsWith('browser_'), output: out } } -// ---- Loops (turn scheduled prompts on/off via a tool, not a UI toggle) ------- - -function runLoopCreate(input: Record, cwd: string): ToolResult { - const name = str(input.name).trim() - const prompt = str(input.prompt).trim() - const interval = Number( - input.interval_minutes ?? input.intervalMinutes ?? input.interval ?? input.minutes - ) - if (!name || !prompt) return { ok: false, output: 'loop_create: needs "name" and "prompt".' } - if (!Number.isFinite(interval) || interval < 1) { - return { ok: false, output: 'loop_create: "interval_minutes" must be a number >= 1.' } - } - const loop = repo.createLoop({ - name, - prompt, - intervalMinutes: Math.floor(interval), - workspacePath: cwd || null - }) - return { - ok: true, - output: `Created loop "${loop.name}" — runs every ${loop.intervalMinutes} min${ - cwd ? ' in this project' : '' - }. It fires shortly and on each interval; pause it with loop_disable.` - } -} - -function runLoopRemove(ref: string): ToolResult { - if (!ref.trim()) return { ok: false, output: 'loop_remove: missing "loop" (a name or id)' } - const loops = repo.listLoops() - const needle = ref.trim().toLowerCase() - const loop = - loops.find((l) => l.id === ref) ?? - loops.find((l) => l.name.toLowerCase() === needle) ?? - loops.find((l) => l.name.toLowerCase().includes(needle)) - if (!loop) return { ok: false, output: `No loop matches "${ref}". Run loop_list to see them.` } - repo.removeLoop(loop.id) - return { ok: true, output: `Removed loop "${loop.name}".` } -} - -function runLoopList(): ToolResult { - const loops = repo.listLoops() - if (loops.length === 0) return { ok: true, output: 'No loops defined.' } - const lines = loops.map( - (l) => - `${l.enabled ? '\u25cf' : '\u25cb'} ${l.name} \u2014 every ${l.intervalMinutes}m (${l.enabled ? 'running' : 'paused'})` - ) - return { ok: true, output: lines.join('\n') } -} - -function runLoopSet(ref: string, enabled: boolean): ToolResult { - const verb = enabled ? 'loop_enable' : 'loop_disable' - if (!ref.trim()) return { ok: false, output: `${verb}: missing "loop" (a name or id)` } - const loops = repo.listLoops() - const needle = ref.trim().toLowerCase() - const loop = - loops.find((l) => l.id === ref) ?? - loops.find((l) => l.name.toLowerCase() === needle) ?? - loops.find((l) => l.name.toLowerCase().includes(needle)) - if (!loop) return { ok: false, output: `No loop matches "${ref}". Run loop_list to see them.` } - repo.setLoopEnabled(loop.id, enabled) - return { ok: true, output: `${enabled ? 'Enabled' : 'Disabled'} loop "${loop.name}".` } -} - // ---- Skills authoring (the agent saving reusable workflows for later) -------- /** diff --git a/src/main/index.ts b/src/main/index.ts index 06d0ee2..5bd6526 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -5,7 +5,8 @@ import icon from '../../resources/icon.png?asset' import macDockIcon from '../../resources/icon-mac.png?asset' import { registerIpc } from './ipc' import { getDb } from './db/database' -import { startLoopScheduler } from './services/loops' +import { startAutomation, stopAutomation } from './services/automation' +import { stopAllTurns } from './services/turn-state' import { listModels } from './services/models' import { backfillUsageFromHistory } from './services/usage' import { listConnectedProviders } from './db/repo' @@ -125,7 +126,7 @@ app.whenReady().then(() => { // and IPC are up so nothing here can delay the first window, and it owns its // own storage - a failure in it can't touch either. initTracking() - startLoopScheduler() + startAutomation() // Sweep tool-output spill files older than the retention window (best-effort). void cleanupToolOutputs() // One-time: seed the usage/cost table from existing message history so the @@ -154,6 +155,8 @@ app.on('window-all-closed', () => { // start tearing down, which gives the final flush a real (if not guaranteed) // window to reach the network. Losing it costs one app_close, nothing more. app.on('before-quit', () => { + stopAutomation() + stopAllTurns() shutdownTracking() }) diff --git a/src/main/ipc/index.ts b/src/main/ipc/index.ts index 4116a02..ca4fc2d 100644 --- a/src/main/ipc/index.ts +++ b/src/main/ipc/index.ts @@ -7,7 +7,6 @@ import { clipboardHasContent, runClipboardAction } from '../services/context-men import type { CookieRow, CreateChatInput, - CreateLoopInput, CreateWorktreeInput, CreateWorktreeResult, ForkChatInput, @@ -94,6 +93,17 @@ import { import { DEFAULT_THEME_ID, type PlatformId, type ResolvedTheme } from '../../shared/theme' import { applyWindowChromeAll } from '../services/window-chrome' import { runSessionTurn } from '../services/session-turn' +import * as bots from '../db/bots' +import { getDb } from '../db/database' +import type { BotJobInput } from '../../shared/bots' +import { + automationSnapshot, + enqueuePrompt, + notifyAutomation, + notifyBots, + wakeAutomation +} from '../services/automation' +import { claimTurn, sessionBusy, stopTurn, resumeQueue } from '../services/turn-state' import { isTrackingEnabled, markActivation, @@ -108,6 +118,7 @@ import { BUNDLE_FILENAME } from '../../shared/portable' /** In-flight streamed completions, keyed by requestId, so they can be aborted. */ const llmControllers = new Map() +const localTurnReleases = new Map void }>() /** * Every abortable piece of model work a SESSION currently owns. @@ -145,6 +156,7 @@ function trackSession(sessionId: string | undefined, controller: AbortController /** Abort everything in flight for a session (its turn, and any compaction). */ function abortSession(sessionId: string): void { + stopTurn(sessionId) for (const controller of sessionControllers.get(sessionId) ?? []) controller.abort() } @@ -304,6 +316,8 @@ export function registerIpc(): void { repo.setChatConfig(id, patch) ) ipcMain.handle(CHANNELS.chatsRemove, (_e, id: string) => { + if (bots.chatBot(id)) throw new Error('Delete bots from their settings, not from a project') + abortSession(id) // Cancel any background subagents this session launched before it's deleted, // so detached work doesn't keep running against a gone parent. cancelSessionBackgroundJobs(id) @@ -704,13 +718,48 @@ export function registerIpc(): void { return applyImport(text) }) - // ---- loops ---- - ipcMain.handle(CHANNELS.loopsList, () => repo.listLoops()) - ipcMain.handle(CHANNELS.loopsCreate, (_e, input: CreateLoopInput) => repo.createLoop(input)) - ipcMain.handle(CHANNELS.loopsSetEnabled, (_e, id: string, enabled: boolean) => - repo.setLoopEnabled(id, enabled) + // ---- bots ---- + ipcMain.handle(CHANNELS.botsList, () => bots.listBots()) + ipcMain.handle(CHANNELS.botsCreate, (_e, username: string) => { + const bot = bots.createBot(username) + notifyBots() + return bot + }) + ipcMain.handle( + CHANNELS.botsUpdate, + (_e, id: string, patch: { username?: string; instructions?: string }) => { + const bot = bots.updateBot(id, patch) + notifyBots() + return bot + } ) - ipcMain.handle(CHANNELS.loopsRemove, (_e, id: string) => repo.removeLoop(id)) + ipcMain.handle(CHANNELS.botsRemove, (_e, id: string) => { + const bot = bots.getBot(id) + if (bot && sessionBusy(bot.chatId)) throw new Error('Stop the bot before deleting it') + if (bot) { + cancelSessionBackgroundJobs(bot.chatId) + endSubagentRuns(bot.chatId) + killSessionBackground(bot.chatId) + browser.disposeSession(bot.chatId) + } + bots.removeBot(id) + notifyBots() + }) + ipcMain.handle(CHANNELS.botsJobs, (_e, botId: string) => bots.listJobs(botId)) + ipcMain.handle(CHANNELS.botsSaveJob, (_e, input: BotJobInput, id?: string) => { + const job = bots.saveJob(input, id) + notifyBots() + return job + }) + ipcMain.handle(CHANNELS.botsRemoveJob, (_e, id: string) => { + const job = bots.listJobs().find((entry) => entry.id === id) + bots.removeJob(id) + notifyBots() + const bot = job && bots.getBot(job.botId) + if (bot) notifyAutomation(bot.chatId) + }) + ipcMain.handle(CHANNELS.automationSnapshot, () => automationSnapshot()) + ipcMain.handle(CHANNELS.automationWake, () => wakeAutomation()) // ---- tools ---- ipcMain.handle( @@ -720,7 +769,16 @@ export function registerIpc(): void { // card and the agent never operate on different trees. const cwd = sessionCwd(sessionId) // Browser & loop tools don't need a workspace; file/bash tools do. - const needsWorkspace = !name.startsWith('browser_') && !name.startsWith('loop_') + const needsWorkspace = [ + 'read', + 'write', + 'edit', + 'glob', + 'grep', + 'list', + 'bash', + 'lsp' + ].includes(name) if (!cwd && needsWorkspace) { return { ok: false, output: 'No workspace is open for this session.' } } @@ -740,22 +798,31 @@ export function registerIpc(): void { ipcMain.handle( CHANNELS.queueAdd, (_e, chatId: string, content: string, images?: QueueImage[]) => { - const item = repo.enqueue(chatId, content, images) + const item = enqueuePrompt(chatId, content, images) remote.notifyQueueChanged() return item } ) ipcMain.handle(CHANNELS.queueRemove, (_e, id: string) => { + const item = getDb().prepare('SELECT chat_id FROM queue WHERE id = ?').get(id) as + | { chat_id: string } + | undefined repo.removeQueueItem(id) remote.notifyQueueChanged() + if (item) notifyAutomation(item.chat_id) }) ipcMain.handle(CHANNELS.queueReorder, (_e, chatId: string, ids: string[]) => { repo.reorderQueue(chatId, ids) remote.notifyQueueChanged() + notifyAutomation(chatId) }) ipcMain.handle(CHANNELS.queueUpdate, (_e, id: string, content: string, images?: QueueImage[]) => { const item = repo.updateQueueItem(id, content, images) remote.notifyQueueChanged() + if (item) { + resumeQueue(item.chatId) + notifyAutomation(item.chatId) + } return item }) @@ -770,7 +837,35 @@ export function registerIpc(): void { // path runs the exact same code. Here we just own the AbortController (for // llm:abort) and stream each event to the renderer that started the turn. ipcMain.handle(CHANNELS.llmStart, async (event, input: LlmStartInput) => { + if (localTurnReleases.has(input.requestId)) + return { ok: false, error: 'Request ID is already in use.' } + if (!repo.getChat(input.sessionId)) return { ok: false, error: 'Session not found.' } const controller = new AbortController() + const release = claimTurn(input.sessionId, controller) + if (!release) + return { ok: false, error: 'This session is already running. Queue your message instead.' } + let abandoned = false + const onDestroyed = (): void => { + abandoned = true + controller.abort() + if (!llmControllers.has(input.requestId)) release() + event.sender.removeListener('destroyed', onDestroyed) + event.sender.removeListener('did-start-loading', onDestroyed) + event.sender.removeListener('render-process-gone', onDestroyed) + localTurnReleases.delete(input.requestId) + } + event.sender.once('destroyed', onDestroyed) + event.sender.once('did-start-loading', onDestroyed) + event.sender.once('render-process-gone', onDestroyed) + localTurnReleases.set(input.requestId, { + senderId: event.sender.id, + release: () => { + event.sender.removeListener('destroyed', onDestroyed) + event.sender.removeListener('did-start-loading', onDestroyed) + event.sender.removeListener('render-process-gone', onDestroyed) + release() + } + }) llmControllers.set(input.requestId, controller) const untrack = trackSession(input.sessionId, controller) // Stop can be pressed in the gap between the renderer asking for the turn @@ -779,6 +874,9 @@ export function registerIpc(): void { if (controller.signal.aborted) { llmControllers.delete(input.requestId) untrack() + release() + localTurnReleases.get(input.requestId)?.release() + localTurnReleases.delete(input.requestId) return { ok: false, error: 'Stopped.' } } // If this session is shared to a phone, relay the turn there too so the phone @@ -801,9 +899,16 @@ export function registerIpc(): void { } finally { llmControllers.delete(input.requestId) untrack() + if (abandoned) release() if (relay) remote.relayLocalTurnEnd(relay) } }) + ipcMain.handle(CHANNELS.llmFinish, (event, requestId: string) => { + const pending = localTurnReleases.get(requestId) + if (!pending || pending.senderId !== event.sender.id || llmControllers.has(requestId)) return + pending.release() + localTurnReleases.delete(requestId) + }) ipcMain.handle(CHANNELS.llmAbort, (_e, requestId: string) => { llmControllers.get(requestId)?.abort() }) @@ -873,11 +978,14 @@ export function registerIpc(): void { // ahead of the turn it makes room for, and is often the longest thing // standing between pressing Stop and anything happening. const controller = new AbortController() + const release = claimTurn(chatId, controller) + if (!release) throw new Error('This session is already running') const untrack = trackSession(chatId, controller) try { return await compactChat(chatId, providerId, model, controller.signal) } finally { untrack() + release() } } ) diff --git a/src/main/services/automation.ts b/src/main/services/automation.ts new file mode 100644 index 0000000..352510c --- /dev/null +++ b/src/main/services/automation.ts @@ -0,0 +1,408 @@ +/** Main-process queue owner. Claims are durable; renderer windows only mirror turns. */ +import { BrowserWindow } from 'electron' +import { randomUUID } from 'node:crypto' +import type { QueueImage, QueueItem, MessagePart } from '../../shared/types' +import type { ChatMessage, RemoteDelta } from '../../shared/api' +import { CHANNELS } from '../../shared/ipc' +import { mentionedBot } from '../../shared/bots' +import { PartsFold, partsToContent } from '../../shared/parts' +import { reconstructTurn } from '../../shared/tool-history' +import { pruneToolMessages, KEEP_RECENT_TOKENS } from '../../shared/context' +import { + resolveSessionConfig, + contextBudgetFor, + clampReasoningEffort +} from '../../shared/session-config' +import { pickDefaultModel } from '../../shared/models' +import * as repo from '../db/repo' +import * as bots from '../db/bots' +import { getDb } from '../db/database' +import { listModels } from './models' +import { compactChat } from './compaction' +import { subagentSnapshot } from './subagent-stream' +import { runSessionTurn } from './session-turn' +import { claimTurn, sessionBusy, queuePaused, resumeQueue } from './turn-state' +import { + relayLocalTurnStart, + relayLocalTurnEvent, + relayLocalTurnEnd, + notifyQueueChanged, + notifyTranscriptChanged +} from './remote' + +let timer: ReturnType | null = null +const live = new Map() + +export function notifyAutomation(chatId: string): void { + for (const win of BrowserWindow.getAllWindows()) { + try { + if (!win.isDestroyed()) win.webContents.send(CHANNELS.automationChanged, chatId) + } catch { + /* window teardown */ + } + } + notifyQueueChanged() +} + +export function notifyBots(): void { + for (const win of BrowserWindow.getAllWindows()) { + try { + if (!win.isDestroyed()) win.webContents.send(CHANNELS.botsChanged) + } catch { + /* window teardown */ + } + } +} + +function emit(delta: RemoteDelta): void { + for (const win of BrowserWindow.getAllWindows()) { + try { + if (!win.isDestroyed()) win.webContents.send(CHANNELS.automationDelta, delta) + } catch { + /* window teardown */ + } + } +} + +export function automationSnapshot(): { sessionId: string; parts: MessagePart[] }[] { + return [...live].map(([sessionId, fold]) => ({ sessionId, parts: fold.parts })) +} + +export function enqueuePrompt( + chatId: string, + content: string, + images?: QueueImage[], + options: { + sourceChatId?: string + replyToChatId?: string + notBefore?: number + hops?: number + continueReply?: boolean + } = {} +): QueueItem { + if (!repo.getChat(chatId)) throw new Error('Session not found') + if (typeof content !== 'string' || (!content.trim() && !images?.length)) + throw new Error('A prompt is required') + if ( + options.hops !== undefined && + (!Number.isSafeInteger(options.hops) || options.hops > 8 || options.hops < 0) + ) { + throw new Error('Handoff limit reached. Ask the user before starting another chain.') + } + if ( + options.notBefore !== undefined && + (!Number.isSafeInteger(options.notBefore) || options.notBefore < 0) + ) { + throw new Error('not_before must be an epoch timestamp in milliseconds') + } + const count = getDb() + .prepare('SELECT COUNT(*) AS n FROM queue WHERE chat_id = ?') + .get(chatId) as { n: number } + if (count.n >= 100) throw new Error('This session already has 100 queued messages') + const item = getDb().transaction(() => { + const item = repo.enqueue(chatId, content.trim(), images) + getDb() + .prepare( + 'UPDATE queue SET source_chat_id = ?, reply_to_chat_id = ?, hops = ?, not_before = ?, continue_reply = ? WHERE id = ?' + ) + .run( + options.sourceChatId ?? null, + options.replyToChatId ?? null, + options.hops ?? 0, + options.notBefore ?? 0, + Number(!!options.continueReply), + item.id + ) + return item + })() + if (!options.sourceChatId) resumeQueue(chatId) + notifyAutomation(chatId) + // Drain on the next event-loop pass, after callers have persisted their own tool result. + if (timer) + setImmediate(() => { + if (timer) wakeAutomation() + }) + return repo.listQueue(chatId).find((entry) => entry.id === item.id)! +} + +export function startAutomation(): void { + if (timer) return + // Never replay uncertain tool side effects automatically after a crash. + getDb() + .prepare( + `UPDATE queue SET state = 'failed', error = 'Interrupted by app shutdown. Edit this message to retry.' WHERE state = 'running'` + ) + .run() + timer = setInterval(wakeAutomation, 1000) + wakeAutomation() +} + +export function stopAutomation(): void { + if (timer) clearInterval(timer) + timer = null +} + +export function wakeAutomation(): void { + try { + const scheduled = bots.enqueueDueJobs() + if (scheduled.length) notifyBots() + for (const id of scheduled) notifyAutomation(id) + const rows = getDb() + .prepare(`SELECT DISTINCT chat_id FROM queue WHERE state = 'pending'`) + .all() as { chat_id: string }[] + for (const { chat_id: chatId } of rows) { + if (live.size >= 4) break + if (sessionBusy(chatId) || queuePaused(chatId) || subagentSnapshot(chatId) !== null) continue + const item = repo.listQueue(chatId)[0] + if (!item || item.state !== 'pending' || (item.notBefore ?? 0) > Date.now()) continue + const target = mentionedBot(item.content, bots.listBots()) + if (target && target.chatId !== chatId) { + try { + getDb().transaction(() => { + const source = repo.getChat(chatId) + const context = repo + .listMessages(chatId) + .slice(-8) + .map((m) => `${m.role}: ${m.content}`) + .join('\n') + .slice(-24000) + const prompt = item.content.replace(/^\s*@[a-z][a-z0-9_-]*[\s,:]*/i, '') || 'Hello' + enqueuePrompt( + target.chatId, + `Request from session ${chatId} (${source?.title ?? ''}).\n\n${prompt}\n\n\n${context}\n`, + item.images, + { sourceChatId: chatId, replyToChatId: chatId, hops: (item.hops ?? 0) + 1 } + ) + repo.addMessage({ + chatId, + role: 'user', + content: item.content, + parts: [ + { type: 'text', text: item.content }, + ...(item.images ?? []).map((image) => ({ type: 'image' as const, ...image })) + ] + }) + repo.removeQueueItem(item.id) + })() + } catch (error) { + getDb() + .prepare(`UPDATE queue SET state = 'failed', error = ? WHERE id = ?`) + .run(String(error), item.id) + } + notifyAutomation(chatId) + notifyTranscriptChanged(chatId) + continue + } + void deliver(item).catch((error) => console.error('[bots] delivery failed', error)) + } + } catch (error) { + console.error('[bots] scheduler failed', error) + } +} + +function history(chatId: string, budget: number, outputReserve: number): ChatMessage[] { + const since = repo.getChat(chatId)?.contextSummaryAt ?? 0 + const groups = repo + .listMessages(chatId) + .filter((m) => (m.role === 'user' || m.role === 'assistant') && m.createdAt > since) + .map(reconstructTurn) + .filter((g) => g.length) + const pruned = pruneToolMessages(groups.flat(), { keepRecentTokens: KEEP_RECENT_TOKENS }) + let index = 0 + const rebuilt = groups.map((g) => g.map(() => pruned[index++])) + const cap = Math.max(2000, budget - outputReserve - 6000) + const kept: ChatMessage[][] = [] + let used = 0 + for (let i = rebuilt.length - 1; i >= 0; i--) { + const tokens = rebuilt[i].reduce( + (sum, m) => + sum + + Math.ceil((m.content.length + JSON.stringify(m.toolCalls ?? []).length) / 4) + + (m.images?.length ?? 0) * 800, + 0 + ) + if (kept.length && used + tokens > cap) break + kept.unshift(rebuilt[i]) + used += tokens + } + const flat = kept.flat() + while (flat.length && flat[0].role !== 'user') flat.shift() + if (flat.at(-1)?.role !== 'user') + flat.push({ role: 'user', content: 'Continue with the pending request.' }) + return flat +} + +async function deliver(item: QueueItem): Promise { + const controller = new AbortController() + const release = claimTurn(item.chatId, controller) + if (!release) return + const claimed = getDb() + .prepare(`UPDATE queue SET state = 'running', error = NULL WHERE id = ? AND state = 'pending'`) + .run(item.id) + if (!claimed.changes) { + release() + return + } + const fold = new PartsFold() + live.set(item.chatId, fold) + emit({ sessionId: item.chatId, kind: 'turn', state: 'running' }) + const relay = relayLocalTurnStart(item.chatId) + try { + const config = resolveSessionConfig(repo.getChat(item.chatId), repo.getSettings()) + const providers = repo.listConnectedProviders().filter((p) => p.enabled) + const provider = config.providerId + ? providers.find((p) => p.id === config.providerId) + : providers[0] + if (!provider) + throw new Error( + 'Connect the provider selected for this session, then edit the queued message to retry.' + ) + const catalog = await listModels(provider.id).catch(() => []) + const model = + (config.providerId === provider.id ? config.model : null) || + provider.defaultModel || + pickDefaultModel(catalog) + if (!model) throw new Error('Select a model for this session, then retry.') + if (controller.signal.aborted) throw new Error('Stopped.') + const previous = getDb().prepare('SELECT message_id FROM queue WHERE id = ?').get(item.id) as + | { message_id: string | null } + | undefined + if (!previous) return + if (!previous.message_id) + getDb().transaction(() => { + const message = repo.addMessage({ + chatId: item.chatId, + role: 'user', + content: item.content, + parts: [ + { type: 'text', text: item.content }, + ...(item.images ?? []).map((image) => ({ type: 'image' as const, ...image })) + ] + }) + getDb().prepare('UPDATE queue SET message_id = ? WHERE id = ?').run(message.id, item.id) + })() + notifyAutomation(item.chatId) + notifyTranscriptChanged(item.chatId) + const info = catalog.find((m) => m.id === model) + const budget = contextBudgetFor(config.contextLimit, info?.contextLimit ?? 128000) + const chat = repo.getChat(item.chatId) + const estimated = repo + .listMessages(item.chatId) + .filter((m) => m.createdAt > (chat?.contextSummaryAt ?? 0)) + .reduce((sum, m) => sum + Math.ceil(m.content.length / 4), 0) + if (estimated > budget * 0.8) + await compactChat(item.chatId, provider.id, model, controller.signal) + if (controller.signal.aborted) throw new Error('Stopped.') + const result = await runSessionTurn( + { + requestId: randomUUID(), + sessionId: item.chatId, + providerId: provider.id, + model, + messages: history(item.chatId, budget, info?.outputLimit ?? 4096), + agentId: config.agentId, + reasoning: info?.reasoning, + reasoningEffort: clampReasoningEffort(config.reasoningEffort, info?.reasoningEfforts), + contextLimit: budget + }, + (event) => { + fold.apply(event) + emit({ sessionId: item.chatId, kind: 'event', event }) + if (relay) relayLocalTurnEvent(relay, event) + }, + controller.signal + ) + const bot = bots.chatBot(item.chatId) + const parts = fold.parts + getDb().transaction(() => { + if (!repo.getChat(item.chatId)) return + if (parts.length) + repo.addMessage({ + chatId: item.chatId, + role: 'assistant', + content: partsToContent(parts), + parts, + botId: bot?.id, + botUsername: bot?.username + }) + if (!result.ok) throw new Error(result.error ?? 'Model request failed') + if ( + item.replyToChatId && + item.replyToChatId !== item.chatId && + repo.getChat(item.replyToChatId) + ) { + // A reply is a transcript result, not a new prompt: never trigger a reply loop. + const text = parts + .filter((p): p is Extract => p.type === 'text') + .map((p) => p.text) + .join('\n') + repo.addMessage({ + chatId: item.replyToChatId, + role: 'assistant', + content: text || partsToContent(parts), + botId: bot?.id, + botUsername: bot?.username + }) + const continuation = getDb() + .prepare('SELECT continue_reply FROM queue WHERE id = ?') + .get(item.id) as { continue_reply: number } | undefined + const pending = getDb() + .prepare('SELECT COUNT(*) AS n FROM queue WHERE chat_id = ?') + .get(item.replyToChatId) as { n: number } + if (continuation?.continue_reply && (item.hops ?? 0) < 8 && pending.n < 100) { + enqueuePrompt( + item.replyToChatId, + `Your delegated request to ${bot ? `@${bot.username}` : `session ${item.chatId}`} completed. Continue the original task if needed, or report the result. Do not reflexively invoke the sender again.\n\n\n${text || partsToContent(parts)}\n`, + undefined, + { sourceChatId: item.chatId, hops: (item.hops ?? 0) + 1 } + ) + } + } + getDb().prepare('DELETE FROM queue WHERE id = ?').run(item.id) + })() + if (bot) notifyBots() + if (item.replyToChatId) { + notifyAutomation(item.replyToChatId) + notifyTranscriptChanged(item.replyToChatId) + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + getDb() + .prepare(`UPDATE queue SET state = 'failed', error = ? WHERE id = ?`) + .run(message, item.id) + if ( + item.replyToChatId && + item.replyToChatId !== item.chatId && + repo.getChat(item.replyToChatId) + ) { + const bot = bots.chatBot(item.chatId) + repo.addMessage({ + chatId: item.replyToChatId, + role: 'assistant', + content: `The delegated request could not finish: ${message}\nThe request remains in ${bot ? `@${bot.username}'s` : "the target session's"} queue for retry or removal.`, + botId: bot?.id, + botUsername: bot?.username + }) + notifyAutomation(item.replyToChatId) + notifyTranscriptChanged(item.replyToChatId) + } + if (fold.parts.length && repo.getChat(item.chatId)) { + const bot = bots.chatBot(item.chatId) + repo.addMessage({ + chatId: item.chatId, + role: 'assistant', + content: partsToContent(fold.parts), + parts: fold.parts, + botId: bot?.id, + botUsername: bot?.username + }) + } + } finally { + live.delete(item.chatId) + release() + if (relay) relayLocalTurnEnd(relay) + notifyTranscriptChanged(item.chatId) + notifyAutomation(item.chatId) + emit({ sessionId: item.chatId, kind: 'turn', state: 'idle' }) + } +} diff --git a/src/main/services/compaction.ts b/src/main/services/compaction.ts index 1d9a5b2..b9480da 100644 --- a/src/main/services/compaction.ts +++ b/src/main/services/compaction.ts @@ -72,7 +72,7 @@ export async function compactChat( // Most recent ~120k chars (older turns matter less if the convo is enormous). const convo = messages - .map((m) => `${m.role.toUpperCase()}: ${flatten(m)}`) + .map((m) => `${m.botUsername ? `@${m.botUsername}` : m.role.toUpperCase()}: ${flatten(m)}`) .join('\n\n') .slice(-120_000) const prior = existing.contextSummary diff --git a/src/main/services/loops.ts b/src/main/services/loops.ts deleted file mode 100644 index 4ad314a..0000000 --- a/src/main/services/loops.ts +++ /dev/null @@ -1,49 +0,0 @@ -/** - * Loop heartbeat scheduler. Every tick it fires any enabled loops whose - * next run is due: it appends the loop's prompt + a heartbeat response to the - * loop's chat, then broadcasts so open windows refresh live. - * - * Each heartbeat uses the real list_sessions data, demonstrating the tools a - * loop will drive once the model + agent loop are wired. - */ -import { BrowserWindow } from 'electron' -import { CHANNELS } from '../../shared/ipc' -import type { Loop } from '../../shared/types' -import * as repo from '../db/repo' - -const CHECK_INTERVAL_MS = 30_000 - -let timer: ReturnType | null = null - -export function startLoopScheduler(): void { - if (timer) return - timer = setInterval(tick, CHECK_INTERVAL_MS) - // Run shortly after startup so freshly created / due loops fire promptly. - setTimeout(tick, 3_000) -} - -export function stopLoopScheduler(): void { - if (timer) clearInterval(timer) - timer = null -} - -function tick(): void { - const now = Date.now() - let due: Loop[] - try { - due = repo.dueLoops(now) - } catch { - return - } - for (const loop of due) { - // Advance the schedule here; the renderer runs the real agent turn on tick. - repo.markLoopRan(loop.id) - broadcast(loop.id) - } -} - -function broadcast(loopId: string): void { - for (const win of BrowserWindow.getAllWindows()) { - win.webContents.send(CHANNELS.loopsTick, loopId) - } -} diff --git a/src/main/services/remote.ts b/src/main/services/remote.ts index 10a7af3..afd1110 100644 --- a/src/main/services/remote.ts +++ b/src/main/services/remote.ts @@ -14,33 +14,16 @@ * put. `start` mints + connects, `stop` tears down + revokes, and `remote:state` * pushes keep the desktop dialog live. */ -import { randomUUID } from 'node:crypto' import { BrowserWindow } from 'electron' import { is } from '@electron-toolkit/utils' import WebSocket from 'ws' import { CHANNELS } from '../../shared/ipc' -import type { - ChatMessage, - LlmEvent, - ModelInfo, - RemoteDelta, - RemotePhase, - RemoteState, - RemoteStartInput -} from '../../shared/api' +import type { LlmEvent, RemotePhase, RemoteState, RemoteStartInput } from '../../shared/api' import type { Message } from '../../shared/types' -import { reconstructTurn } from '../../shared/tool-history' -import { PartsFold, partsToContent } from '../../shared/parts' -import { pruneToolMessages, KEEP_RECENT_TOKENS } from '../../shared/context' -import { - clampReasoningEffort, - contextBudgetFor, - resolveSessionConfig -} from '../../shared/session-config' +import { PartsFold } from '../../shared/parts' import * as repo from '../db/repo' -import { listModels } from './models' -import { pickDefaultModel } from '../../shared/models' -import { runSessionTurn } from './session-turn' +import { enqueuePrompt } from './automation' +import { stopTurn } from './turn-state' import { track, trackFeature } from './track' import { sessionCwd } from './workspace' import { @@ -63,9 +46,6 @@ const WS_BASE = HTTP_BASE.replace(/^http/, 'ws') /** Reconnect backoff (ms) after an unexpected host-socket drop; then give up. */ const RECONNECT_DELAYS_MS = [1_000, 2_000, 5_000, 10_000, 15_000] -/** Reserve for the system prompt (prepended inside runAgentTurn) in the window cut. */ -const SYSTEM_RESERVE_TOKENS = 6_000 - /** The mint response from `POST /api/remote/sessions`. */ interface MintResponse { brokerId: string @@ -89,12 +69,6 @@ interface Share { guests: number phase: RemotePhase error?: string - /** Abort handles for in-flight remote turns, keyed by sessionId (one per session). */ - turns: Map - /** Sessions with an in-flight *desktop-driven* turn (via `llm:start`), tracked - * through the relay hooks so a phone prompt queues behind a local turn instead - * of starting a second concurrent one on the same session. */ - localTurns: Set /** Live parts accumulators for in-flight turns, so a guest that joins/switches * mid-turn can be seeded with the reply-so-far (keyed by sessionId). */ liveTurns: Map @@ -135,19 +109,6 @@ function broadcast(): void { } } -/** - * Push a phone-driven turn's live event to every open window so the desktop - * mirrors the reply token-by-token (the local `llm:delta` twin for remote turns). - * Share-bound: a no-op once `active` is no longer the current share, so a stale - * turn can't leak deltas into a replaced session. - */ -function broadcastDeltaFor(active: Share, payload: RemoteDelta): void { - if (share !== active) return - for (const win of BrowserWindow.getAllWindows()) { - if (!win.isDestroyed()) win.webContents.send(CHANNELS.remoteDelta, payload) - } -} - /** Bump the revision + push. Called on any status change or shared-session activity. */ function bump(patch?: Partial>): void { if (!share) return @@ -314,257 +275,18 @@ function switchSession(sessionId: string): void { bump() } -// --- Turn assembly (mirrors the renderer's buildChatMessages) -------------- - -/** - * Rebuild the chat-completion history for a session within the context budget — - * the main-process twin of the renderer's `buildChatMessages`. The real system - * prompt (and any compaction summary) is prepended inside `runAgentTurn`, so it's - * only reserved for here, not materialized. - */ -function buildRemoteMessages( - sessionId: string, - contextBudget: number, - outputReserve: number -): ChatMessage[] { - const chat = repo.getChat(sessionId) - const since = chat?.contextSummaryAt ?? 0 - // Group each persisted turn so the window cut can never split an assistant's - // tool_calls from the matching role:'tool' results (which would 400 providers). - const groups = repo - .listMessages(sessionId) - .filter((m) => (m.role === 'user' || m.role === 'assistant') && m.createdAt > since) - .map(reconstructTurn) - .filter((g) => g.length > 0) - - // Prune older tool outputs to a head/tail preview before the cut, then zip back - // into groups so tool_calls stay paired with their results. - const flatAll = groups.flat() - const prunedFlat = pruneToolMessages(flatAll, { keepRecentTokens: KEEP_RECENT_TOKENS }) - let pk = 0 - const prunedGroups = groups.map((g) => g.map(() => prunedFlat[pk++])) - - const cap = Math.max(2000, contextBudget - outputReserve - SYSTEM_RESERVE_TOKENS) - const estimate = (m: ChatMessage): number => - Math.ceil((m.content.length + (m.toolCalls ? JSON.stringify(m.toolCalls).length : 0)) / 4) + - (m.images?.length ?? 0) * 800 - const groupTokens = (g: ChatMessage[]): number => g.reduce((n, m) => n + estimate(m), 0) - - const kept: ChatMessage[][] = [] - let used = 0 - for (let i = prunedGroups.length - 1; i >= 0; i--) { - const tokens = groupTokens(prunedGroups[i]) - if (used + tokens > cap && kept.length > 0) break - kept.unshift(prunedGroups[i]) - used += tokens - } - const flat = kept.flat() - // Normalize the leading edge to a user message (Anthropic requires it and a - // dangling assistant/tool would orphan a tool_use). The current prompt is at - // the tail, so this only trims stale boundary turns. - while (flat.length && flat[0].role !== 'user') flat.shift() - return flat -} - -// --- The crux: run a guest's prompt exactly like a local one --------------- - -/** - * Is a turn already in flight for this session — from *either* end? A phone turn - * lives in `turns`; a desktop turn (via `llm:start`) is tracked in `localTurns` - * through the relay hooks. Both queue gates consult this so the phone and desktop - * share one FIFO instead of each only respecting its own in-flight turn. - */ -function isSessionBusy(active: Share, sessionId: string): boolean { - return active.turns.has(sessionId) || active.localTurns.has(sessionId) -} - -/** - * Handle a prompt typed on the phone. Mirrors the desktop's `submit`: if a turn - * is already running for this session — from the phone OR the desktop — or prompts - * are already queued, append it to the shared FIFO instead of starting a second - * turn. Otherwise run it now. The queue is the same persisted `repo` queue the - * desktop uses, so the pending list stays identical on both ends. - */ +/** Phone prompts join the same durable queue as desktop and scheduled prompts. */ async function handlePrompt(sessionId: string, text: string): Promise { const active = share if (!active || !text.trim()) return - if (!repo.getChat(sessionId)) { - sendFrame({ t: 'error', message: 'That session no longer exists.' }) - return - } - // A turn is already running for this session (either end) — or prompts are - // already queued — so queue this one (FIFO), mirror the updated queue to the - // phone(s), and nudge the desktop so its queue view refreshes too. Draining - // happens automatically when the current turn ends. (Matches the desktop's - // single gate: queue while a turn is in flight or a backlog already exists.) - if (isSessionBusy(active, sessionId) || repo.listQueue(sessionId).length > 0) { - repo.enqueue(sessionId, text.trim()) - sendQueue(sessionId) - bumpFor(active) - return - } - await runTurn(active, sessionId, text.trim(), false) -} - -/** - * Run one turn for a guest's prompt, exactly like a local one, then drain the - * next queued prompt (if any). Called by `handlePrompt` for a fresh prompt and - * by `drainRemoteQueue` for each dequeued one — neither re-checks the busy guard, - * so a drained prompt actually runs rather than re-queuing behind itself. - * - * `announce` is true for a drained queue item: the phone never echoed it (it only - * had it in the pending list), so the host sends the user text on `turn:running` - * for the phone to show its bubble. A direct phone send echoes locally, so it's - * false there to avoid a double bubble. - */ -async function runTurn( - active: Share, - sessionId: string, - text: string, - announce: boolean -): Promise { - // Serialize turns *per session*: claim the slot synchronously so two quick - // prompts can't start concurrent turns on the same session (mirrors the - // renderer's guard). Different sessions can still run independently. Also - // yields to an in-flight desktop turn (tracked in `localTurns`). - if (isSessionBusy(active, sessionId)) { - // A turn slipped in first — fall back to queuing so nothing is lost. - repo.enqueue(sessionId, text) - sendQueue(sessionId) - bumpFor(active) - return - } - const controller = new AbortController() - active.turns.set(sessionId, controller) - // Register the live accumulator up-front so a guest that joins/switches during - // this turn (even mid provider-resolution) is seeded with the reply-so-far. - const acc = new PartsFold() - active.liveTurns.set(sessionId, acc) - try { - // Persist the user's message as if typed locally, then nudge the desktop. - repo.addMessage({ chatId: sessionId, role: 'user', content: text }) - bumpFor(active) - // Announce the prompt text for a drained queue item so the phone shows its - // bubble (a direct send already echoed it locally). `sendQueue` above already - // removed it from the pending list, so it moves cleanly from queue → turn. - sendFrameFor(active, { - t: 'turn', - sessionId, - state: 'running', - userText: announce ? text : undefined - }) - // Mirror the turn start to the desktop so it opens a live bubble now (the - // user message was just persisted + bumped above; the reply streams next). - broadcastDeltaFor(active, { sessionId, kind: 'turn', state: 'running' }) - - // Resolve THIS SESSION's config, exactly as the renderer does - through the - // one shared resolver, so a phone turn runs on the model the desktop shows - // for that session rather than whatever was last picked globally. - const settings = repo.getSettings() - const config = resolveSessionConfig(repo.getChat(sessionId), settings) - const providers = repo.listConnectedProviders() - const provider = providers.find((p) => p.id === config.providerId) ?? providers[0] ?? null - if (!provider) { - sendFrameFor(active, { t: 'error', message: 'No provider is connected on the desktop.' }) - return - } - // Fetch the catalog first so we can pick the provider's latest tool-capable - // model when none was explicitly chosen (mirrors the renderer). A hardcoded - // id may not exist on this provider, which would 404 the first phone turn. - let catalog: ModelInfo[] = [] - try { - catalog = await listModels(provider.id) - } catch { - // Offline model catalog — fall back to conservative defaults below. - } - const model = - (config.providerId === provider.id ? config.model : null) || - provider.defaultModel || - pickDefaultModel(catalog) || - (provider.id === 'github-copilot' ? 'gpt-4o' : 'gpt-4o-mini') - const info = catalog.find((m) => m.id === model) - const modelContext = info?.contextLimit ?? 128_000 - const contextBudget = contextBudgetFor(config.contextLimit, modelContext) - const messages = buildRemoteMessages(sessionId, contextBudget, info?.outputLimit ?? 4096) - - const result = await runSessionTurn( - { - requestId: randomUUID(), - sessionId, - providerId: provider.id, - model, - messages, - // The session's own mode: a session left in Plan mode stays read-only - // when it is driven from the phone. - agentId: config.agentId, - reasoning: info?.reasoning ?? false, - // Same clamp as the desktop send path: the session's effort is sticky, - // the model's ladder is not, and an unsupported level 400s the turn. - reasoningEffort: clampReasoningEffort(config.reasoningEffort, info?.reasoningEfforts), - contextLimit: contextBudget - }, - (event) => { - acc.apply(event) - sendFrameFor(active, { t: 'delta', sessionId, event }) - // Fan the same event to the desktop renderer so the PC streams the reply - // live, exactly like a local turn (the phone and desktop stay in lockstep). - broadcastDeltaFor(active, { sessionId, kind: 'event', event }) - }, - controller.signal - ) - - // Persist the assistant reply (mirrors the renderer's post-stream persistence) - // so the desktop transcript updates and the next turn keeps the context. - const parts = acc.parts - if (!result.ok && !controller.signal.aborted) { - parts.push({ type: 'text', text: `_\u26a0 ${result.error ?? 'Model request failed.'}_` }) - } - if (parts.length) { - repo.addMessage({ - chatId: sessionId, - role: 'assistant', - content: partsToContent(parts), - parts - }) - } - bumpFor(active) - } finally { - // Always release the turn slot (even on an unexpected throw) so future - // prompts aren't permanently rejected; only clear if we still own it. - if (active.turns.get(sessionId) === controller) active.turns.delete(sessionId) - if (active.liveTurns.get(sessionId) === acc) active.liveTurns.delete(sessionId) - sendFrameFor(active, { t: 'turn', sessionId, state: 'idle' }) - // Drop the desktop's live bubble; the persisted reply (bumped above) is - // reconciled from disk by the renderer's mirror, so this hands off cleanly. - broadcastDeltaFor(active, { sessionId, kind: 'turn', state: 'idle' }) - // A turn stopped by the user shouldn't auto-run the backlog — a phone abort - // leaves the queued prompts in place (the phone can drain them with a fresh - // send, mirroring the desktop's Stop). Otherwise drain the next queued prompt. - if (!controller.signal.aborted) void drainRemoteQueue(active, sessionId) - } -} - -/** - * Run the next pending prompt for a session, chaining until the queue is empty. - * Each dequeued prompt runs through `runTurn`, which drains again when it ends — - * so the whole backlog streams to the phone one turn at a time. A no-op if the - * share was replaced, a turn is already running, or the queue is empty. - */ -async function drainRemoteQueue(active: Share, sessionId: string): Promise { - if (share !== active || isSessionBusy(active, sessionId)) return - const items = repo.listQueue(sessionId) - if (items.length === 0) { + enqueuePrompt(sessionId, text) sendQueue(sessionId) - return + bumpFor(active) + } catch (error) { + sendFrame({ t: 'error', message: error instanceof Error ? error.message : String(error) }) } - const next = items[0] - repo.removeQueueItem(next.id) - sendQueue(sessionId) - bumpFor(active) - await runTurn(active, sessionId, next.content, true) } - /** * Re-broadcast the shared queue to the phone(s) after a *desktop-side* change * (the renderer added/removed/reordered a queued prompt). Called from the queue @@ -574,6 +296,11 @@ export function notifyQueueChanged(): void { if (share) sendQueue(share.currentSessionId) } +/** Reconcile queued turns and out-of-band bot replies with the phone transcript. */ +export function notifyTranscriptChanged(sessionId: string): void { + if (share?.currentSessionId === sessionId) sendSnapshot(sessionId) +} + // --- Relaying a *desktop-driven* turn to the phone ------------------------- /** Opaque handle threading one local turn's relay state through the IPC layer. */ @@ -598,9 +325,6 @@ export interface LocalTurnRelay { export function relayLocalTurnStart(sessionId: string, userText?: string): LocalTurnRelay | null { const active = share if (!active) return null - // Mark the session busy so a phone prompt queues behind this desktop turn - // instead of starting a second concurrent one (the shared busy gate). - active.localTurns.add(sessionId) const acc = new PartsFold() active.liveTurns.set(sessionId, acc) // `userText` mirrors the drained-queue announce path: the phone never echoed a @@ -627,15 +351,10 @@ export function relayLocalTurnEvent(relay: LocalTurnRelay, event: LlmEvent): voi * `finishTurn`); the phone's own authoritative snapshot reconciles it on the next * switch/reconnect. * - * We deliberately do NOT drain the shared queue here: the desktop renderer's - * `finishTurn` already drains it after a local turn (via `drainQueue`), so a - * prompt the phone queued behind this turn runs as the desktop's next send. - * Kicking `drainRemoteQueue` too would race that renderer drain into two - * concurrent turns on the same session. + * Queue ownership belongs to automation; relay teardown never drains or cancels it. */ export function relayLocalTurnEnd(relay: LocalTurnRelay): void { const { active, sessionId, acc } = relay - active.localTurns.delete(sessionId) if (active.liveTurns.get(sessionId) === acc) active.liveTurns.delete(sessionId) sendFrameFor(active, { t: 'turn', sessionId, state: 'idle' }) } @@ -735,14 +454,18 @@ function onFrame(raw: string): void { break } case 'abort': { - share.turns.get(share.currentSessionId)?.abort() + stopTurn(share.currentSessionId) break } case 'dequeue': { // Phone tapped × on a queued prompt — drop it from the shared queue and // re-broadcast so both ends update. `bump` refreshes the desktop's view. if (typeof frame.id === 'string') { - repo.removeQueueItem(frame.id) + try { + repo.removeQueueItem(frame.id) + } catch (error) { + sendFrame({ t: 'error', message: error instanceof Error ? error.message : String(error) }) + } sendQueue(share.currentSessionId) bump() } @@ -787,9 +510,6 @@ function teardown(): void { clearTimeout(active.reconnectTimer) active.reconnectTimer = null } - for (const controller of active.turns.values()) controller.abort() - active.turns.clear() - active.localTurns.clear() active.liveTurns.clear() const sock = active.socket active.socket = null @@ -889,8 +609,6 @@ async function startInternal(input: RemoteStartInput): Promise { socket: null, guests: 0, phase: 'starting', - turns: new Map(), - localTurns: new Set(), liveTurns: new Map(), reconnectAttempts: 0, reconnectTimer: null, diff --git a/src/main/services/session-turn.ts b/src/main/services/session-turn.ts index 79df555..f780664 100644 --- a/src/main/services/session-turn.ts +++ b/src/main/services/session-turn.ts @@ -198,6 +198,7 @@ async function runTurn( signal, emit }) + if (signal.aborted) return { ok: false, error: 'Stopped.' } // The turn's subagents are one-shot — drop any with nothing queued so they // don't linger in the sidebar after the work is done. Spared: sub sessions // with a still-running background task (Phase 11), one still streaming, and diff --git a/src/main/services/turn-state.ts b/src/main/services/turn-state.ts new file mode 100644 index 0000000..2983de8 --- /dev/null +++ b/src/main/services/turn-state.ts @@ -0,0 +1,31 @@ +/** Shared ownership across desktop, phone, and scheduled turns. */ +const active = new Map() +const paused = new Set() + +export function sessionBusy(id: string): boolean { + return active.has(id) +} + +export function claimTurn(id: string, controller: AbortController): (() => void) | null { + if (active.has(id)) return null + active.set(id, controller) + return () => { + if (active.get(id) === controller) active.delete(id) + } +} + +export function stopTurn(id: string): void { + paused.add(id) + active.get(id)?.abort() +} + +export function queuePaused(id: string): boolean { + return paused.has(id) +} +export function resumeQueue(id: string): void { + paused.delete(id) +} + +export function stopAllTurns(): void { + for (const controller of active.values()) controller.abort() +} diff --git a/src/main/services/workspace.ts b/src/main/services/workspace.ts index 173b52b..d2407f6 100644 --- a/src/main/services/workspace.ts +++ b/src/main/services/workspace.ts @@ -11,7 +11,8 @@ * consumer must agree, or a session will read from one tree and write to * another. */ -import { existsSync, readdirSync } from 'node:fs' +import { existsSync, readdirSync, mkdirSync } from 'node:fs' +import { app } from 'electron' import path from 'node:path' import * as repo from '../db/repo' import { resolveWorktreeCwd } from '../../shared/workspace' @@ -179,6 +180,11 @@ export function sessionCwd(chatId: string): string { continue } const workspacePath = chat.workspacePath + if (chat.kind === 'bot' && !workspacePath) { + const home = path.join(app.getPath('userData'), 'bots', chat.id) + mkdirSync(home, { recursive: true }) + return home + } if (!workspacePath) return '' if (!chat.worktreePath) return workspacePath return resolveWorktreeCwd( diff --git a/src/preload/index.ts b/src/preload/index.ts index c671b0b..0b7ce42 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -21,6 +21,35 @@ import type { ResolvedTheme } from '../shared/theme' * to an ipcMain.handle channel registered in src/main/ipc/index.ts. */ const roxy: RoxyApi = { + bots: { + list: () => ipcRenderer.invoke(CHANNELS.botsList), + create: (username) => ipcRenderer.invoke(CHANNELS.botsCreate, username), + update: (id, patch) => ipcRenderer.invoke(CHANNELS.botsUpdate, id, patch), + remove: (id) => ipcRenderer.invoke(CHANNELS.botsRemove, id), + jobs: (botId) => ipcRenderer.invoke(CHANNELS.botsJobs, botId), + saveJob: (input, id) => ipcRenderer.invoke(CHANNELS.botsSaveJob, input, id), + removeJob: (id) => ipcRenderer.invoke(CHANNELS.botsRemoveJob, id), + onChanged: (callback) => { + const handler = (): void => callback() + ipcRenderer.on(CHANNELS.botsChanged, handler) + return () => ipcRenderer.removeListener(CHANNELS.botsChanged, handler) + } + }, + automation: { + snapshot: () => ipcRenderer.invoke(CHANNELS.automationSnapshot), + wake: () => ipcRenderer.invoke(CHANNELS.automationWake), + onChanged: (callback) => { + const handler = (_event: Electron.IpcRendererEvent, chatId: string): void => callback(chatId) + ipcRenderer.on(CHANNELS.automationChanged, handler) + return () => ipcRenderer.removeListener(CHANNELS.automationChanged, handler) + }, + onDelta: (callback) => { + const handler = (_event: Electron.IpcRendererEvent, payload: RemoteDelta): void => + callback(payload) + ipcRenderer.on(CHANNELS.automationDelta, handler) + return () => ipcRenderer.removeListener(CHANNELS.automationDelta, handler) + } + }, settings: { getAll: () => ipcRenderer.invoke(CHANNELS.settingsGetAll), setActiveProvider: (providerId, model) => @@ -144,17 +173,6 @@ const roxy: RoxyApi = { export: () => ipcRenderer.invoke(CHANNELS.configExport), import: () => ipcRenderer.invoke(CHANNELS.configImport) }, - loops: { - list: () => ipcRenderer.invoke(CHANNELS.loopsList), - create: (input) => ipcRenderer.invoke(CHANNELS.loopsCreate, input), - setEnabled: (id, enabled) => ipcRenderer.invoke(CHANNELS.loopsSetEnabled, id, enabled), - remove: (id) => ipcRenderer.invoke(CHANNELS.loopsRemove, id), - onTick: (callback) => { - const handler = (_event: Electron.IpcRendererEvent, loopId: string): void => callback(loopId) - ipcRenderer.on(CHANNELS.loopsTick, handler) - return () => ipcRenderer.removeListener(CHANNELS.loopsTick, handler) - } - }, tools: { run: (sessionId, name, input) => ipcRenderer.invoke(CHANNELS.toolsRun, sessionId, name, input), cancel: (callId) => ipcRenderer.invoke(CHANNELS.toolsCancel, callId) @@ -175,6 +193,7 @@ const roxy: RoxyApi = { }, llm: { start: (input) => ipcRenderer.invoke(CHANNELS.llmStart, input), + finish: (requestId) => ipcRenderer.invoke(CHANNELS.llmFinish, requestId), abort: (requestId) => ipcRenderer.invoke(CHANNELS.llmAbort, requestId), abortSession: (sessionId) => ipcRenderer.invoke(CHANNELS.llmAbortSession, sessionId), onDelta: (callback) => { diff --git a/src/renderer/src/canvas/CanvasTranscript.tsx b/src/renderer/src/canvas/CanvasTranscript.tsx index 6171642..aa3b5ad 100644 --- a/src/renderer/src/canvas/CanvasTranscript.tsx +++ b/src/renderer/src/canvas/CanvasTranscript.tsx @@ -6,6 +6,8 @@ import { transcriptCache, layoutTranscript } from './transcript' import type { HitAction } from './scene' import { promptEntries } from './prompt-history' import roxyLogo from '../assets/roxy.png' +import { useRoxyStore } from '../lib/store' +import { botAvatarUrl } from '../components/BotAvatar' export type { CanvasProbe } from './CanvasSurface' @@ -55,6 +57,12 @@ export function CanvasTranscript({ const [logo, setLogo] = useState(() => decodedLogo) const [clock, setClock] = useState(0) const prompts = useMemo(() => promptEntries(messages), [messages]) + const bots = useRoxyStore((s) => s.bots) + const ownBot = bots.find((bot) => bot.chatId === chatId) + // Preserve message references where possible. A rename changes only the + // identity cache key, not every part in a long transcript. + const identities = bots.map((bot) => `${bot.id}:${bot.username}`).join('|') + const identityRef = useRef('') useEffect(() => () => cache.detach(), [cache]) @@ -83,6 +91,11 @@ export function CanvasTranscript({ const buildScene = useCallback( (context: CanvasLayoutContext) => { void clock + const identityKey = `${chatId}:${identities}` + if (identityRef.current !== identityKey) { + cache.clear() + identityRef.current = identityKey + } cache.prune(messages) if (logo) context.view.images.set('__roxy__', logo) return layoutTranscript( @@ -90,6 +103,9 @@ export function CanvasTranscript({ ...context, messages, streaming, + botUsername: ownBot?.username, + bots, + botAvatar: botAvatarUrl, canCancel: (part) => { if (part.tool === 'task') return Boolean(part.subChatId) return ( @@ -101,7 +117,7 @@ export function CanvasTranscript({ cache ) }, - [messages, streaming, clock, logo, cache] + [messages, streaming, clock, logo, cache, bots, ownBot?.username, chatId, identities] ) const onAction = (action: HitAction): void => { diff --git a/src/renderer/src/canvas/transcript-window.ts b/src/renderer/src/canvas/transcript-window.ts index c02ab98..59a2120 100644 --- a/src/renderer/src/canvas/transcript-window.ts +++ b/src/renderer/src/canvas/transcript-window.ts @@ -4,6 +4,7 @@ import type { Block, Scene } from './scene' import { FONT_SIZE, SPACE } from './metrics' import { layoutMessageHeader, + messageBotUsername, layoutParts, layoutUserBody, partsText, @@ -269,7 +270,7 @@ export class TranscriptWindow { .map(([id, state]) => `${id}:${state.left}:${state.top}`) .join(',') const live = message.id === '__streaming__' - const key = `${format}:${opened}:${diffs}:${scrolls}:${live && item.part === message.parts.length - 1}:${part?.type === 'tool' ? part.state : ''}` + const key = `${format}:${item.kind === 'header' ? (messageBotUsername(input, message) ?? '') : ''}:${opened}:${diffs}:${scrolls}:${live && item.part === message.parts.length - 1}:${part?.type === 'tool' ? part.state : ''}` const hit = this.entries.get(item.id) let block: Block if (!live && hit?.key === key && sameSource(hit.source, source)) { @@ -279,9 +280,18 @@ export class TranscriptWindow { } else { const builder = new Builder(input.metrics, input.theme, { value: 0 }, input.t) let height: number - if (item.kind === 'header') - height = layoutMessageHeader(builder, message.role === 'user', x, 0, width).y - else if (item.kind === 'user') + if (item.kind === 'header') { + const username = messageBotUsername(input, message) + height = layoutMessageHeader( + builder, + message.role === 'user', + x, + 0, + width, + username, + username ? input.botAvatar?.(username) : undefined + ).y + } else if (item.kind === 'user') height = layoutUserBody(builder, message.parts, bodyX, 0, bodyWidth) else if (item.kind === 'end') height = diff --git a/src/renderer/src/canvas/transcript.ts b/src/renderer/src/canvas/transcript.ts index 0e48acf..0656774 100644 --- a/src/renderer/src/canvas/transcript.ts +++ b/src/renderer/src/canvas/transcript.ts @@ -24,8 +24,12 @@ import { layoutMarkdown, layoutPlainText } from './prose' import { layoutToolCard, type ToolCardInput } from './tool-card' import { PROMPT_GUTTER } from './prompt-history' import { TranscriptWindow } from './transcript-window' +import type { Bot } from '@shared/bots' export interface LayoutInput { + botUsername?: string + bots?: Bot[] + botAvatar?: (username: string) => string messages: Message[] /** The live turn's parts, or null when nothing is streaming. */ streaming: MessagePart[] | null @@ -146,7 +150,16 @@ function layoutMessage( streaming = false ): Block { const builder = new Builder(input.metrics, input.theme, counter, input.t) - const body = layoutMessageHeader(builder, message.role === 'user', x, y, width) + const username = messageBotUsername(input, message) + const body = layoutMessageHeader( + builder, + message.role === 'user', + x, + y, + width, + username, + username ? input.botAvatar?.(username) : undefined + ) let cursor = body.y if (message.role === 'user') { cursor += layoutUserBody(builder, message.parts, body.x, cursor, body.width) @@ -166,12 +179,23 @@ function layoutMessage( return { ...builder.finish(message.id, y, height), copyText: () => partsText(message.parts) } } +export function messageBotUsername(input: LayoutInput, message: Message): string | undefined { + if (message.role !== 'assistant') return undefined + return ( + input.bots?.find((bot) => bot.id === message.botId)?.username ?? + message.botUsername ?? + input.botUsername + ) +} + export function layoutMessageHeader( builder: Builder, isUser: boolean, x: number, y: number, - width: number + width: number, + botUsername?: string, + botAvatarSrc?: string ): { x: number; y: number; width: number } { const palette = builder.palette const top = y + SPACE.messagePadY @@ -198,8 +222,8 @@ export function layoutMessageHeader( y: avatarY, w: SPACE.avatar, h: SPACE.avatar, - src: '__roxy__', - radius: SPACE.radiusLg, + src: botAvatarSrc ?? '__roxy__', + radius: botUsername ? SPACE.avatar / 2 : SPACE.radiusLg, border: palette.border }) } @@ -208,7 +232,7 @@ export function layoutMessageHeader( builder.text( bodyX, top, - builder.t(isUser ? 'transcript.you' : 'transcript.assistant'), + botUsername ? `@${botUsername}` : builder.t(isUser ? 'transcript.you' : 'transcript.assistant'), nameFont, palette.textMuted ) diff --git a/src/renderer/src/components/BotAvatar.tsx b/src/renderer/src/components/BotAvatar.tsx new file mode 100644 index 0000000..10a1fc1 --- /dev/null +++ b/src/renderer/src/components/BotAvatar.tsx @@ -0,0 +1,45 @@ +import { Facehash, FACES, stringHash } from 'facehash' +import { renderToStaticMarkup } from 'react-dom/server' + +const COLORS = ['#b5a1e8', '#e6ae96', '#94cbbb', '#9dbde4', '#d9c783'] + +/** One identity across the sidebar, mentions, and canvas transcript. */ +export function BotAvatar({ + username, + size = 32 +}: { + username: string + size?: number +}): JSX.Element { + return ( + + ) +} + +const avatarUrls = new Map() + +/** Use Facehash's actual SVG face on canvas, not a DOM-only alternate transcript. */ +export function botAvatarUrl(username: string): string { + const cached = avatarUrls.get(username) + if (cached) return cached + const hash = stringHash(username) + const Face = FACES[hash % FACES.length] + const eyes = renderToStaticMarkup().replace( + '${eyes}` + const url = `data:image/svg+xml,${encodeURIComponent(svg)}` + avatarUrls.set(username, url) + return url +} diff --git a/src/renderer/src/components/BotSettingsPane.tsx b/src/renderer/src/components/BotSettingsPane.tsx new file mode 100644 index 0000000..a8abe1e --- /dev/null +++ b/src/renderer/src/components/BotSettingsPane.tsx @@ -0,0 +1,499 @@ +import { useEffect, useState } from 'react' +import { useTranslation } from 'react-i18next' +import { Pencil, Plus, Trash2, X } from 'lucide-react' +import type { Bot, BotJob, BotSchedule } from '@shared/bots' +import { api } from '../lib/api' +import { useRoxyStore } from '../lib/store' +import { BotAvatar } from './BotAvatar' +import { Button, Input, Textarea } from './ui' + +const fieldClass = 'flex flex-col gap-1.5 text-xs text-text-muted' +const selectClass = + 'h-9 w-full rounded-lg border border-border bg-surface-2 px-2 text-sm text-text outline-none focus:border-accent' +const message = (error: unknown): string => (error instanceof Error ? error.message : String(error)) + +/** Settings stay alongside the existing chat; jobs are configuration, not a second chat stack. */ +export function BotSettingsPane({ bot, onClose }: { bot: Bot; onClose: () => void }): JSX.Element { + const { t } = useTranslation() + const refreshBots = useRoxyStore((s) => s.refreshBots) + const removeBot = useRoxyStore((s) => s.removeBot) + const [username, setUsername] = useState(bot.username) + const [instructions, setInstructions] = useState(bot.instructions) + const [jobs, setJobs] = useState([]) + const [loading, setLoading] = useState(true) + const [error, setError] = useState('') + const [busy, setBusy] = useState(false) + const [saved, setSaved] = useState(false) + const [confirmDelete, setConfirmDelete] = useState(false) + const [editingJob, setEditingJob] = useState(null) + + useEffect(() => { + let live = true + const reload = async (): Promise => { + try { + const next = await api.bots.jobs(bot.id) + if (live) setJobs(next) + } catch (e) { + if (live) setError(message(e)) + } finally { + if (live) setLoading(false) + } + } + void reload() + const off = api.bots.onChanged(() => void reload()) + return () => { + live = false + off() + } + }, [bot.id]) + + const run = async (action: () => Promise): Promise => { + if (busy) return + setBusy(true) + setError('') + try { + await action() + } catch (e) { + setError(message(e)) + } finally { + setBusy(false) + } + } + + return ( +