diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..907bd9b --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,154 @@ +# AGENTS.md + +## Overview + +This is **Antigravity Agent** — an autonomous AI agent platform built with a Rust backend and a vanilla JS frontend. Users create "tasks" (goals), and the system dispatches an LLM-powered agent to accomplish them. Tasks can be run on-demand or on a cron schedule. Each execution produces logs and a final answer, all persisted to PostgreSQL. + +## Architecture + +``` +┌─────────────────────────────────────────────────────────┐ +│ Frontend (Vite + Vanilla JS) │ +│ - OAuth login via Authentik │ +│ - Dashboard: recent runs, task list, logs, answers │ +│ - Polls /api/tasks every 3s for live updates │ +└──────────────────────┬──────────────────────────────────┘ + │ /api/* +┌──────────────────────▼──────────────────────────────────┐ +│ Axum HTTP Server (src/server/) │ +│ - CORS, CSP, rate limiting (1 MB body) │ +│ - Cookie-based + Bearer token auth │ +│ - Routes: tasks CRUD, runs, auth (callback/refresh) │ +├─────────────────────────────────────────────────────────┤ +│ Domain Layer (src/domain/) │ +│ ├─ agent/ Agent loop, LLM API client, tool defs │ +│ ├─ auth JwksVerifier, Authenticator (OIDC) │ +│ └─ tasks Task execution, run management │ +├─────────────────────────────────────────────────────────┤ +│ Scheduler (src/scheduler.rs) │ +│ - tokio-cron-scheduler for recurring task execution │ +├─────────────────────────────────────────────────────────┤ +│ Entities (src/entities/) │ +│ - task, task_run (SeaORM models) │ +├─────────────────────────────────────────────────────────┤ +│ PostgreSQL │ +│ - Migrations managed via sea-orm-migration │ +└─────────────────────────────────────────────────────────┘ +``` + +## Project Structure + +``` +bot/ +├── Cargo.toml # Workspace root (members: ".", "migration") +├── src/ +│ ├── main.rs # Entrypoint: tracing init → server::start() +│ ├── config.rs # Config struct loaded from env vars +│ ├── error.rs # AppError enum (thiserror) → Axum responses +│ ├── scheduler.rs # Cron job scheduler (wraps tokio-cron-scheduler) +│ ├── tests.rs # Unit tests for error handling & config +│ ├── domain/ +│ │ ├── agent/ +│ │ │ ├── mod.rs # Agent struct: agentic loop with turn/time limits +│ │ │ ├── api.rs # LLM request/response types, Tavily search client +│ │ │ └── tools.rs # Tool definitions (google_search, finish) & dispatch +│ │ ├── auth.rs # JwksVerifier (RSA/JWKS), Authenticator (code exchange, refresh) +│ │ └── tasks.rs # Task execution logic, response DTOs +│ ├── entities/ +│ │ ├── task.rs # SeaORM entity: tasks table +│ │ └── task_run.rs # SeaORM entity: task_runs table (belongs_to task) +│ └── server/ +│ ├── mod.rs # App bootstrap: DB, scheduler, auth, router, CORS +│ ├── auth.rs # Auth routes & AuthenticatedUser extractor +│ └── tasks.rs # Task CRUD & run endpoints +├── migration/ +│ └── src/ # SeaORM migrations (tasks, answer col, runs table, cron col) +├── frontend/ +│ ├── index.html # SPA shell with glassmorphism dark theme +│ ├── src/ +│ │ ├── main.js # All app logic: auth flow, task/run rendering, polling +│ │ └── style.css # Styles +│ ├── vite.config.js # Dev proxy: /api → localhost:3000 +│ └── package.json # Deps: vite, marked, dompurify +└── .forgejo/workflows/ + └── pipeline.yaml # CI: build frontend + cargo build → deploy via systemd +``` + +## Agent System + +The agent (`src/domain/agent/`) is a turn-based autonomous loop: + +1. A system prompt is injected with the current date and instructions not to ask the user for clarification. +2. The user's goal is sent as the initial message. +3. Each turn calls the **Kimi K2.5** model via the Zen API (`https://opencode.ai/zen/v1/chat/completions`). +4. The model can invoke tools: + - **`google_search`** — web search via the Tavily API. + - **`finish`** — signals completion and provides the final answer. +5. Tool results are appended to the conversation and the loop continues. +6. The loop terminates when `finish` is called, the turn limit is hit (`AGENT_MAX_TURNS`, default 20), or the time limit expires (`AGENT_MAX_DURATION_SECS`, default 120s). + +All turns and tool calls are logged. The final answer (if any) and the full log are persisted to the `task_runs` table. + +## Environment Variables + +| Variable | Required | Default | Description | +|---------------------------|----------|---------|----------------------------------------------| +| `DATABASE_URL` | ✅ | — | PostgreSQL connection string | +| `PORT` | | `3000` | HTTP server port | +| `ZEN_API_KEY` | | — | API key for Zen/Kimi LLM | +| `TAVILY_API_KEY` | | — | API key for Tavily web search | +| `AUTHENTIK_ISSUER` | ✅ | — | OIDC issuer URL (Authentik) | +| `AUTHENTIK_CLIENT_ID` | ✅ | — | OAuth client ID | +| `AUTHENTIK_CLIENT_SECRET` | ✅ | — | OAuth client secret | +| `CORS_ALLOWED_ORIGINS` | | — | Comma-separated allowed origins (or mirror) | +| `COOKIE_SECURE` | | `false` | Set `true` for HTTPS-only cookies | +| `AGENT_MAX_TURNS` | | `20` | Max LLM turns per agent run | +| `AGENT_MAX_DURATION_SECS` | | `120` | Max wall-clock seconds per agent run | + +## API Routes + +All task/run routes require authentication (cookie or Bearer token). + +| Method | Path | Description | +|--------|------------------------|-----------------------------------| +| GET | `/api/tasks` | List all tasks with their runs | +| POST | `/api/tasks` | Create a new task | +| GET | `/api/tasks/:id` | Get a single task with runs | +| PUT | `/api/tasks/:id` | Update task goal/cron schedule | +| POST | `/api/tasks/:id/runs` | Trigger a manual re-run | +| GET | `/api/runs/recent` | Latest 50 runs across all tasks | +| GET | `/api/auth/session` | Check current session | +| GET | `/api/auth/callback` | OAuth code → token exchange | +| POST | `/api/auth/refresh` | Refresh access token | +| POST | `/api/auth/logout` | Clear auth cookies | + +## Database Schema + +**`tasks`** — `id` (UUID PK), `goal` (text), `cron` (text, nullable), `created_at` (timestamptz) + +**`task_runs`** — `id` (UUID PK), `task_id` (FK → tasks, cascade delete), `status` (text), `logs` (text), `answer` (text, nullable), `created_at` (timestamptz) + +Migrations are in `migration/src/` and run automatically on startup via `Migrator::up()`. + +## Development + +```bash +# Backend (from repo root) +cargo run # requires DATABASE_URL + Authentik vars + +# Frontend (from frontend/) +npm install +npm run dev # Vite dev server on :5173, proxies /api to :3000 + +# Tests +cargo test +``` + +## CI / Deployment + +The Forgejo Actions pipeline (`.forgejo/workflows/pipeline.yaml`) triggers on release publish: + +1. Builds the frontend (`npm install && npm run build`). +2. Builds the Rust binary (`cargo build -r`). +3. Uploads the binary as a release asset. +4. Deploys to the host: copies binary + frontend dist, restarts the `bot` systemd user service. diff --git a/Cargo.lock b/Cargo.lock index 73d8ffb..8030dcb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -161,6 +161,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", + "base64", "bytes", "futures-util", "http", @@ -179,8 +180,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -336,7 +339,7 @@ dependencies = [ "sea-orm-migration", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "tokio", "tokio-cron-scheduler", "tower-http 0.5.2", @@ -667,6 +670,12 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "data-encoding" +version = "2.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7a1e2f27636f116493b8b860f5546edb47c8d8f8ea73e1d2a20be88e28d1fea" + [[package]] name = "der" version = "0.7.10" @@ -2505,7 +2514,7 @@ dependencies = [ "serde_json", "sqlx", "strum 0.26.3", - "thiserror", + "thiserror 2.0.18", "time", "tracing", "url", @@ -2602,7 +2611,7 @@ dependencies = [ "proc-macro2", "quote", "syn 2.0.114", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -2814,7 +2823,7 @@ checksum = "297f631f50729c8c99b84667867963997ec0b50f32b2a7dbcab828ef0541e8bb" dependencies = [ "num-bigint", "num-traits", - "thiserror", + "thiserror 2.0.18", "time", ] @@ -2912,7 +2921,7 @@ dependencies = [ "serde_json", "sha2", "smallvec", - "thiserror", + "thiserror 2.0.18", "time", "tokio", "tokio-stream", @@ -3000,7 +3009,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "time", "tracing", "uuid", @@ -3043,7 +3052,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "time", "tracing", "uuid", @@ -3070,7 +3079,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror", + "thiserror 2.0.18", "time", "tracing", "url", @@ -3221,13 +3230,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", ] [[package]] @@ -3381,6 +3410,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -3554,6 +3595,24 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "typenum" version = "1.19.0" @@ -3611,6 +3670,12 @@ dependencies = [ "serde", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" diff --git a/Cargo.toml b/Cargo.toml index 6d3bcb2..e7a13dc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ tokio = { version = "1", features = ["full"] } reqwest = { version = "0.12", features = ["json"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" -axum = "0.7" +axum = { version = "0.7", features = ["ws"] } tower-http = { version = "0.5", features = ["cors", "set-header", "limit"] } sea-orm = { version = "1.1", features = ["sqlx-postgres", "runtime-tokio-rustls", "macros"] } sea-orm-migration = "1.1" diff --git a/frontend/index.html b/frontend/index.html index ea8c9bd..5be239e 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -63,22 +63,56 @@
-

Recent Activity

-

Track latest agent executions across all directives.

+

Perspective

+

Overview of recent agent directives and live communication.

+ +
+ + +
-
- - - - - - - - - - - -
DirectiveStatusDate
+ +
+
+
+
+

Neural Link

+ +
+
+ +
+

Establish connection with the neural assistant.

+
+
+
+ + +
+
+
+ +
+
+ + + + + + + + + + + +
DirectiveStatusDate
+
+
diff --git a/frontend/src/main.js b/frontend/src/main.js index 64ef57e..91e3ffd 100644 --- a/frontend/src/main.js +++ b/frontend/src/main.js @@ -16,7 +16,9 @@ const state = { selectedRunId: null, currentView: 'dashboard', // 'dashboard' or 'task' isEditing: false, - isAuthenticated: false + isAuthenticated: false, + chatMessages: [], + activeDashboardTab: 'chat' // 'chat' or 'activity' }; // DOM elements const loginOverlay = document.getElementById('login-overlay'); @@ -53,6 +55,11 @@ const schedulePresets = document.getElementById('schedule-presets'); const toggleCustomCronBtn = document.getElementById('toggle-custom-cron'); const customCronContainer = document.getElementById('custom-cron-container'); const presetBtns = document.querySelectorAll('.btn-preset'); +const chatMessagesEl = document.getElementById('chat-messages'); +const chatForm = document.getElementById('chat-form'); +const chatInput = document.getElementById('chat-input'); +const clearChatBtn = document.getElementById('clear-chat-btn'); +const chatSendBtn = document.getElementById('chat-send-btn'); function updateState(newState) { Object.assign(state, newState); @@ -89,6 +96,7 @@ function renderApp() { if (state.currentView === 'dashboard') { fetchRecentRuns(); + renderChat(); } else if (state.selectedTaskId) { const task = state.tasks.find(t => t.id === state.selectedTaskId); if (task) { @@ -264,6 +272,28 @@ function showDashboard() { renderTaskList(); fetchRecentRuns(); + renderDashboardTabs(); +} + +function renderDashboardTabs() { + const tabs = document.querySelectorAll('.tab-btn'); + const panels = document.querySelectorAll('.tab-panel'); + + tabs.forEach(tab => { + if (tab.dataset.tab === state.activeDashboardTab) { + tab.classList.add('active'); + } else { + tab.classList.remove('active'); + } + }); + + panels.forEach(panel => { + if (panel.id === `${state.activeDashboardTab}-tab-panel`) { + panel.classList.add('active'); + } else { + panel.classList.remove('active'); + } + }); } function renderDashboard(recentRuns) { @@ -354,6 +384,62 @@ function escapeHtml(text) { return div.innerHTML; } +function renderChat() { + if (!chatMessagesEl) return; + + if (state.chatMessages.length === 0) { + chatMessagesEl.innerHTML = ` +
+

Start a conversation with the assistant.

+
+ `; + return; + } + + chatMessagesEl.innerHTML = state.chatMessages + .map(msg => ` +
+ ${DOMPurify.sanitize(marked.parse(msg.content || ''))} +
+ `) + .join(''); + + chatMessagesEl.scrollTop = chatMessagesEl.scrollHeight; +} + +async function sendChatMessage(text) { + const userMessage = { role: 'user', content: text }; + state.chatMessages.push(userMessage); + renderChat(); + + chatInput.value = ''; + chatInput.disabled = true; + chatSendBtn.disabled = true; + + try { + const response = await fetchWithAuth(`${API_URL}/chat`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ messages: state.chatMessages }) + }); + + if (!response.ok) throw new Error('Chat API failed'); + + const result = await response.json(); + state.chatMessages.push(result.message); + renderChat(); + } catch (error) { + console.error('Chat error:', error); + showToast('Failed to get chat response.', 'error'); + state.chatMessages.push({ role: 'assistant', content: 'Sorry, I encountered an error. Please try again.' }); + renderChat(); + } finally { + chatInput.disabled = false; + chatSendBtn.disabled = false; + chatInput.focus(); + } +} + // Event Listeners rerunBtn.addEventListener('click', async () => { if (!state.selectedTaskId) return; @@ -447,6 +533,19 @@ toggleCustomCronBtn.addEventListener('click', () => { customCronContainer.classList.toggle('hidden'); }); +chatForm.addEventListener('submit', (e) => { + e.preventDefault(); + const text = chatInput.value.trim(); + if (text) { + sendChatMessage(text); + } +}); + +clearChatBtn.addEventListener('click', () => { + state.chatMessages = []; + renderChat(); +}); + cronInput.addEventListener('input', () => { // If user types manually, update presets active state updateScheduleUI(cronInput.value); @@ -511,17 +610,86 @@ async function checkSession() { return false; } -let isPolling = false; -async function startAutoRefresh() { - setInterval(async () => { - if (isPolling) return; - isPolling = true; +let socket = null; +let reconnectDelay = 1000; + +function connectWebSocket() { + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const wsUrl = `${protocol}//${window.location.host}/api/ws`; + + console.log('Connecting to WebSocket:', wsUrl); + socket = new WebSocket(wsUrl); + + socket.onopen = () => { + console.log('WebSocket connected'); + reconnectDelay = 1000; + // Initial fetch to sync state + fetchTasks(); + }; + + socket.onmessage = (event) => { try { - await fetchTasks(); - } finally { - isPolling = false; + const { type, data } = JSON.parse(event.data); + console.log('WebSocket event:', type, data); + + switch (type) { + case 'TaskCreated': + state.tasks.unshift(data); + renderApp(); + showToast('New task created', 'success'); + break; + case 'TaskUpdated': + case 'RunFinished': + const index = state.tasks.findIndex(t => t.id === data.id); + if (index !== -1) { + const wasSelected = state.selectedTaskId === data.id; + state.tasks[index] = data; + if (wasSelected) { + // Update selected run if we were following latest + const wasFollowingLatest = state.selectedRunId === state.tasks[index].runs?.[1]?.id || !state.selectedRunId; + if (wasFollowingLatest && data.runs && data.runs.length > 0) { + state.selectedRunId = data.runs[0].id; + } + } + } else { + state.tasks.unshift(data); + } + renderApp(); + if (type === 'RunFinished') { + showToast(`Task run completed: ${data.goal}`, 'info'); + } + break; + case 'RunStarted': + const taskIndex = state.tasks.findIndex(t => t.id === data.task_id); + if (taskIndex !== -1) { + // We don't have the full task update here, but we can update status + // For simplicity, we just trigger a fetch or wait for RunFinished + // But let's at least show it's running in the UI if selected + if (state.tasks[taskIndex].runs) { + // Prepend a dummy run or just fetch + fetchTasks(); + } + } + showToast(`Task started: ${data.goal}`, 'info'); + break; + } + } catch (e) { + console.error('Error handling WebSocket message:', e); } - }, 3000); + }; + + socket.onclose = () => { + console.log('WebSocket disconnected. Reconnecting...'); + setTimeout(() => { + reconnectDelay = Math.min(reconnectDelay * 1.5, 30000); + connectWebSocket(); + }, reconnectDelay); + }; + + socket.onerror = (error) => { + console.error('WebSocket error:', error); + socket.close(); + }; } async function showLogin() { @@ -578,6 +746,13 @@ logoutBtn.addEventListener('click', () => { logout(); }); +document.querySelectorAll('.tab-btn').forEach(btn => { + btn.addEventListener('click', () => { + state.activeDashboardTab = btn.dataset.tab; + renderDashboardTabs(); + }); +}); + async function initializeApp() { const hasSession = await checkSession(); @@ -585,7 +760,7 @@ async function initializeApp() { appEl.classList.remove('hidden'); loginOverlay.classList.add('hidden'); await fetchTasks(); - startAutoRefresh(); + connectWebSocket(); } else { showLogin(); } diff --git a/frontend/src/style.css b/frontend/src/style.css index 0761f29..f34da31 100644 --- a/frontend/src/style.css +++ b/frontend/src/style.css @@ -88,9 +88,15 @@ body { .dashboard-content { flex: 1; overflow-y: auto; - border-radius: 12px; + border-radius: 16px; display: flex; flex-direction: column; + width: 100%; + max-width: 900px; + background: rgba(255, 255, 255, 0.02); + box-shadow: 0 10px 40px rgba(0, 0, 0, 0.3); + border: 1px solid var(--glass-border); + min-height: 0; } .activity-table { @@ -124,6 +130,158 @@ body { background: rgba(255, 255, 255, 0.03); } +.dashboard-content-grid { + display: flex; + flex-direction: column; + align-items: center; + gap: 32px; + flex: 1; + min-height: 0; + width: 100%; +} + +.dashboard-tabs { + display: flex; + gap: 12px; + margin-top: 24px; +} + +.tab-btn { + background: rgba(255, 255, 255, 0.03); + border: 1px solid var(--glass-border); + color: var(--text-dim); + padding: 10px 20px; + border-radius: 100px; + font-size: 14px; + font-weight: 600; + cursor: pointer; + transition: var(--transition); + display: flex; + align-items: center; + gap: 8px; +} + +.tab-btn:hover { + background: rgba(255, 255, 255, 0.08); + color: var(--text-main); +} + +.tab-btn.active { + background: var(--primary); + color: white; + border-color: var(--primary); + box-shadow: 0 4px 15px var(--primary-glow); +} + +.tab-panel { + display: none; + width: 100%; + flex-direction: column; + align-items: center; + flex: 1; + min-height: 0; +} + +.tab-panel.active { + display: flex; +} + +.chat-container { + display: flex; + flex-direction: column; + border-radius: 16px; + overflow: hidden; + background: rgba(255, 255, 255, 0.02); + width: 100%; + max-width: 900px; + flex: 1; + min-height: 0; + box-shadow: 0 10px 40px rgba(0, 0, 0, 0.3); + border: 1px solid var(--glass-border); +} + +.chat-header { + padding: 16px 20px; + background: rgba(255, 255, 255, 0.03); + border-bottom: 1px solid var(--glass-border); + display: flex; + justify-content: space-between; + align-items: center; +} + +.chat-header h3 { + font-size: 14px; + font-weight: 600; + color: var(--text-dim); + text-transform: uppercase; + letter-spacing: 0.5px; +} + +.chat-messages { + flex: 1; + padding: 20px; + overflow-y: auto; + display: flex; + flex-direction: column; + gap: 12px; +} + +.chat-empty-state { + margin: auto; + text-align: center; + color: var(--text-dim); + font-style: italic; + font-size: 13px; + opacity: 0.6; +} + +.chat-message { + max-width: 80%; + padding: 12px 18px; + border-radius: 16px; + font-size: 15px; + line-height: 1.5; + word-wrap: break-word; +} + +.chat-message.user { + align-self: flex-end; + background: var(--primary); + color: white; + border-bottom-right-radius: 4px; +} + +.chat-message.assistant { + align-self: flex-start; + background: rgba(255, 255, 255, 0.08); + color: var(--text-main); + border-bottom-left-radius: 4px; +} + +.chat-form { + padding: 16px; + background: rgba(255, 255, 255, 0.02); + border-top: 1px solid var(--glass-border); + display: flex; + gap: 10px; +} + +.chat-form input { + flex: 1; + background: rgba(0, 0, 0, 0.2); + border: 1px solid var(--glass-border); + border-radius: 12px; + padding: 12px 20px; + color: var(--text-main); + font-size: 15px; + outline: none; + transition: var(--transition); +} + +.chat-form input:focus { + border-color: var(--primary); +} + .activity-table .status-badge { display: inline-block; } @@ -796,9 +954,9 @@ textarea:focus { } .btn-sm { - padding: 6px 12px; + padding: 6px 16px; font-size: 12px; - width: 100%; + width: auto; justify-content: center; } diff --git a/frontend/vite.config.js b/frontend/vite.config.js index 7ce721b..a71c6f4 100644 --- a/frontend/vite.config.js +++ b/frontend/vite.config.js @@ -6,6 +6,7 @@ export default defineConfig({ '/api': { target: 'http://localhost:3000', changeOrigin: true, + ws: true, } } } diff --git a/src/domain/agent/api.rs b/src/domain/agent/api.rs index 2d2ae37..f886c34 100644 --- a/src/domain/agent/api.rs +++ b/src/domain/agent/api.rs @@ -3,7 +3,6 @@ use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct Message { pub role: String, - #[serde(skip_serializing_if = "Option::is_none")] pub content: Option, #[serde(skip_serializing_if = "Option::is_none")] pub tool_calls: Option>, @@ -29,6 +28,7 @@ pub struct FunctionCall { pub struct ChatRequest { pub model: String, pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] pub tools: Option>, } @@ -72,6 +72,7 @@ pub async fn perform_search( query: &str, api_key: &str, ) -> Result> { + tracing::info!(query = %query, "Performing Tavily web search"); let client = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(30)) .build()?; @@ -94,6 +95,7 @@ pub async fn perform_search( } let search_data: TavilyResponse = response.json().await?; + tracing::info!("Web search yielded {} results", search_data.results.len()); let mut results_text = String::new(); for (i, result) in search_data.results.iter().enumerate() { diff --git a/src/domain/agent/mod.rs b/src/domain/agent/mod.rs index 9f6933f..1b6dde8 100644 --- a/src/domain/agent/mod.rs +++ b/src/domain/agent/mod.rs @@ -2,16 +2,18 @@ pub mod api; pub mod tools; use chrono::Utc; +use sea_orm::DatabaseConnection; use std::time::{Duration, Instant}; use self::api::{ChatRequest, ChatResponse, Message, Tool}; pub struct Agent { + db: DatabaseConnection, client: reqwest::Client, url: String, zen_api_key: Option, tavily_api_key: Option, - messages: Vec, + pub messages: Vec, tools: Option>, logs: String, answer: Option, @@ -21,6 +23,7 @@ use crate::error::{AppError, AppResult}; impl Agent { pub fn new( + db: DatabaseConnection, zen_api_key: Option, tavily_api_key: Option, initial_message: String, @@ -47,14 +50,24 @@ impl Agent { }, ]; + Self::with_messages(db, zen_api_key, tavily_api_key, messages) + } + + pub fn with_messages( + db: DatabaseConnection, + zen_api_key: Option, + tavily_api_key: Option, + messages: Vec, + ) -> AppResult { let tools = Some(tools::get_tools()); let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(60)) + .timeout(std::time::Duration::from_secs(120)) .build() .map_err(|e| AppError::Internal(format!("Failed to build HTTP client: {}", e)))?; Ok(Self { + db, client, url: "https://opencode.ai/zen/v1/chat/completions".to_string(), zen_api_key, @@ -101,6 +114,35 @@ impl Agent { turns, current_role )); + let assistant_message = self.execute_turn().await?; + + if let Some(tool_calls) = &assistant_message.tool_calls { + if tool_calls.iter().any(|tc| tc.function.name == "answer") { + finished = true; + } + } + + if self.answer.is_some() { + finished = true; + } + } + + self.log("\n--- Execution Finished ---"); + Ok((self.logs.clone(), self.answer.clone())) + } + + pub async fn execute_turn(&mut self) -> AppResult { + let max_sub_turns = 10; + let mut sub_turns = 0; + + loop { + sub_turns += 1; + if sub_turns > max_sub_turns { + return Err(AppError::Internal( + "Interaction cycle turn limit exceeded".into(), + )); + } + let chat_response = self.call_llm().await?; let assistant_message = chat_response .choices @@ -117,20 +159,24 @@ impl Agent { } } - if let Some(tool_calls) = assistant_message.tool_calls { + if let Some(tool_calls) = &assistant_message.tool_calls { + let mut is_final_cycle = false; + let mut final_answer = None; + for tool_call in tool_calls { self.log(&format!("Calling tool: {}", tool_call.function.name)); let (tool_message, is_final, tool_answer) = - tools::handle_tool_call(&tool_call, &self.tavily_api_key) + tools::handle_tool_call(tool_call, &self.tavily_api_key, &self.db) .await .map_err(|e| { AppError::Internal(format!("Tool execution failed: {}", e)) })?; if let Some(ans) = tool_answer { - self.answer = Some(ans); - self.log("Task marked as finished by tool."); + self.answer = Some(ans.clone()); + final_answer = Some(ans); + self.log("Interaction marked as finished by tool."); } if let Some(content) = &tool_message.content { @@ -139,20 +185,24 @@ impl Agent { self.messages.push(tool_message); if is_final { - finished = true; + is_final_cycle = true; } } - } else if assistant_message.content.is_some() { - // If assistant just talked without tools, we might be stuck or finished. - // But typically we expect a 'finish' tool call. - self.log("Assistant responded without tool calls."); - // For now we continue unless the assistant explicitly uses a tool to finish, - // or we could add heuristic here if needed. - } - } - self.log("\n--- Execution Finished ---"); - Ok((self.logs.clone(), self.answer.clone())) + if is_final_cycle { + return Ok(Message { + role: "assistant".to_string(), + content: final_answer.or(assistant_message.content), + tool_calls: None, + tool_call_id: None, + }); + } + + continue; + } + + return Ok(assistant_message); + } } async fn call_llm(&self) -> AppResult { @@ -168,23 +218,46 @@ impl Agent { request_builder = request_builder.header("Authorization", format!("Bearer {}", key)); } - let response = request_builder.send().await.map_err(AppError::Network)?; + let start = std::time::Instant::now(); + let response = request_builder.send().await.map_err(|e| { + let duration = start.elapsed(); + let is_timeout = e.is_timeout(); + tracing::error!( + "Network error after {:?} during LLM call (Timeout: {}): {:?}", + duration, + is_timeout, + e + ); + AppError::Network(e) + })?; + + let duration = start.elapsed(); + tracing::info!("LLM request completed in {:?}", duration); if !response.status().is_success() { let status = response.status(); - let error_text = response + let body_text = response .text() .await - .unwrap_or_else(|_| "Unknown error".into()); - return Err(AppError::Internal(format!( - "API request failed: {} - {}", - status, error_text - ))); + .unwrap_or_else(|_| "Unknown body".into()); + let err = format!("API request failed: {} - {}", status, body_text); + tracing::error!("{}", err); + return Err(AppError::Internal(err)); } - response - .json() - .await - .map_err(|e| AppError::Internal(format!("Failed to parse LLM response: {}", e))) + let response_text = response.text().await.map_err(|e| { + let err = format!("Failed to read response text: {}", e); + tracing::error!("{}", err); + AppError::Internal(err) + })?; + + serde_json::from_str(&response_text).map_err(|e| { + let err = format!( + "Failed to parse LLM response: {} | Raw Body: {}", + e, response_text + ); + tracing::error!("{}", err); + AppError::Internal(err) + }) } } diff --git a/src/domain/agent/tools.rs b/src/domain/agent/tools.rs index 663b1b1..bbdff4c 100644 --- a/src/domain/agent/tools.rs +++ b/src/domain/agent/tools.rs @@ -1,5 +1,9 @@ use super::api::{self, FunctionDefinition, Message, Tool, ToolCall}; +use sea_orm::{ + ColumnTrait, Condition, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder, QuerySelect, +}; use serde::Deserialize; +use uuid::Uuid; #[derive(Deserialize)] struct GoogleSearchArgs { @@ -11,6 +15,13 @@ struct FinishArgs { result: String, } +#[derive(Deserialize)] +struct ListRunsArgs { + from: Option, + to: Option, + task_ids: Option>, +} + pub fn get_tools() -> Vec { vec![ Tool { @@ -47,12 +58,51 @@ pub fn get_tools() -> Vec { }), }, }, + Tool { + tool_type: "function".to_string(), + function: FunctionDefinition { + name: "list_tasks".to_string(), + description: "List all existing tasks and their goals".to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": {}, + }), + }, + }, + Tool { + tool_type: "function".to_string(), + function: FunctionDefinition { + name: "list_runs".to_string(), + description: "List task runs with optional filters for date and task IDs" + .to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": { + "from": { + "type": "string", + "description": "ISO 8601 date string to filter runs from" + }, + "to": { + "type": "string", + "description": "ISO 8601 date string to filter runs to" + }, + "task_ids": { + "type": "array", + "items": { "type": "string", "format": "uuid" }, + "description": "List of task IDs to filter runs for" + }, + + }, + }), + }, + }, ] } pub async fn handle_tool_call( tool_call: &ToolCall, tavily_api_key: &Option, + db: &DatabaseConnection, ) -> Result<(Message, bool, Option), Box> { let mut answer = None; let name = &tool_call.function.name; @@ -76,6 +126,78 @@ pub async fn handle_tool_call( answer = Some(result.clone()); (result.clone(), true) + } else if name == "list_tasks" { + tracing::info!("Listing tasks from database"); + use crate::entities::task; + let tasks = task::Entity::find() + .order_by_desc(task::Column::CreatedAt) + .all(db) + .await?; + + let mut out = String::from("Tasks:\n"); + for t in tasks { + out.push_str(&format!("- ID: {}, Goal: {}\n", t.id, t.goal)); + } + (out, false) + } else if name == "list_runs" { + use crate::entities::task_run; + tracing::info!( + "Listing runs from database, args: {}", + tool_call.function.arguments + ); + let args: ListRunsArgs = serde_json::from_str(&tool_call.function.arguments)?; + + tracing::info!( + "Listing runs from database with filters: from={:?}, to={:?}, task_ids={:?}", + args.from, + args.to, + args.task_ids + ); + + let query = task_run::Entity::find(); + + let mut condition = Condition::all(); + + if let Some(from_str) = args.from { + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&from_str) { + condition = condition.add(task_run::Column::CreatedAt.gte(dt)); + } + } + + if let Some(to_str) = args.to { + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&to_str) { + condition = condition.add(task_run::Column::CreatedAt.lte(dt)); + } + } + + if let Some(id_strs) = args.task_ids { + let mut valid_ids = Vec::new(); + for id_str in id_strs { + match id_str.parse::() { + Ok(uuid) => valid_ids.push(uuid), + Err(e) => tracing::warn!("Skipping invalid UUID '{}' from LLM: {}", id_str, e), + } + } + if !valid_ids.is_empty() { + condition = condition.add(task_run::Column::TaskId.is_in(valid_ids)); + } + } + + let runs = query + .filter(condition) + .order_by_desc(task_run::Column::CreatedAt) + .limit(20) + .all(db) + .await?; + + let mut out = String::from("Recent Runs:\n"); + for r in runs { + out.push_str(&format!( + "- ID: {}, Task ID: {}, Status: {}, Created At: {}, Answer: {:?}\n", + r.id, r.task_id, r.status, r.created_at, r.answer + )); + } + (out, false) } else { (format!("Error: Unknown tool {}", name), false) }; diff --git a/src/domain/tasks.rs b/src/domain/tasks.rs index 5092091..6e7962d 100644 --- a/src/domain/tasks.rs +++ b/src/domain/tasks.rs @@ -22,7 +22,7 @@ pub struct UpdateTaskRequest { pub cron: Option, } -#[derive(Serialize)] +#[derive(Serialize, Clone, Debug)] pub struct TaskResponse { pub id: Uuid, pub goal: String, @@ -31,7 +31,7 @@ pub struct TaskResponse { pub runs: Vec, } -#[derive(Serialize)] +#[derive(Serialize, Clone, Debug)] pub struct TaskRunResponse { pub id: Uuid, pub status: String, @@ -40,7 +40,7 @@ pub struct TaskRunResponse { pub created_at: chrono::DateTime, } -#[derive(Serialize)] +#[derive(Serialize, Clone, Debug)] pub struct RecentRunResponse { pub id: Uuid, pub task_id: Uuid, @@ -75,7 +75,20 @@ pub async fn execute_agent_run( .await .map_err(crate::error::AppError::Database)?; + let _ = _scheduler + .tx + .send(crate::server::notifications::WsEvent::RunStarted( + RecentRunResponse { + id: run_id, + task_id, + goal: goal.clone(), + status: "running".to_string(), + created_at: Utc::now().into(), + }, + )); + let mut agent = Agent::new( + db.clone(), config.zen_api_key.clone(), config.tavily_api_key.clone(), goal.clone(), @@ -112,7 +125,14 @@ pub async fn execute_agent_run( .await .map_err(crate::error::AppError::Database)?; - get_task_inner(task_id, db).await + let task_response = get_task_inner(task_id, db).await?; + let _ = _scheduler + .tx + .send(crate::server::notifications::WsEvent::RunFinished( + task_response.clone(), + )); + + Ok(task_response) } pub async fn get_task_inner(id: Uuid, db: &DatabaseConnection) -> AppResult { diff --git a/src/scheduler.rs b/src/scheduler.rs index 2ee8a01..9aad91b 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -13,12 +13,14 @@ pub struct Scheduler { db: DatabaseConnection, tasks_to_jobs: DashMap, config: Arc, + pub tx: tokio::sync::broadcast::Sender, } impl Scheduler { pub async fn new( db: DatabaseConnection, config: Arc, + tx: tokio::sync::broadcast::Sender, ) -> AppResult { let scheduler = JobScheduler::new() .await @@ -32,6 +34,7 @@ impl Scheduler { db, tasks_to_jobs: DashMap::new(), config, + tx, }) } @@ -43,12 +46,14 @@ impl Scheduler { let db = self.db.clone(); let config = self.config.clone(); + let tx = self.tx.clone(); let job = Job::new_async(cron_expr, move |_uuid, _l| { let db = db.clone(); let config = config.clone(); + let tx = tx.clone(); Box::pin(async move { - if let Err(e) = Self::run_task(db, config, task_id).await { + if let Err(e) = Self::run_task(db, config, tx, task_id).await { tracing::error!("Error in scheduled task {}: {}", task_id, e); } }) @@ -81,6 +86,7 @@ impl Scheduler { async fn run_task( db: DatabaseConnection, config: Arc, + tx: tokio::sync::broadcast::Sender, task_id: Uuid, ) -> AppResult<()> { let task = Task::find_by_id(task_id) @@ -105,8 +111,19 @@ impl Scheduler { use sea_orm::ActiveModelTrait; run.insert(&db).await.map_err(AppError::Database)?; + let _ = tx.send(crate::server::notifications::WsEvent::RunStarted( + crate::domain::tasks::RecentRunResponse { + id: run_id, + task_id, + goal: task.goal.clone(), + status: "running".to_string(), + created_at: chrono::Utc::now().into(), + }, + )); + // Start agent in background let mut agent = Agent::new( + db.clone(), config.zen_api_key.clone(), config.tavily_api_key.clone(), task.goal.clone(), @@ -142,6 +159,12 @@ impl Scheduler { e ); } + + if let Ok(task_response) = crate::domain::tasks::get_task_inner(task_id, &db).await { + let _ = tx.send(crate::server::notifications::WsEvent::RunFinished( + task_response, + )); + } }); Ok(()) diff --git a/src/server/chat.rs b/src/server/chat.rs new file mode 100644 index 0000000..5163275 --- /dev/null +++ b/src/server/chat.rs @@ -0,0 +1,57 @@ +use crate::domain::agent::api::Message; +use crate::error::AppError; +use crate::server::AppState; +use axum::{Json, extract::State}; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +#[derive(Debug, Deserialize)] +pub struct ChatPayload { + pub messages: Vec, +} + +#[derive(Debug, Serialize)] +pub struct ChatResult { + pub message: Message, +} + +pub async fn chat_handler( + State(state): State>, + Json(payload): Json, +) -> Result, AppError> { + let msg_count = payload.messages.len(); + tracing::info!("Received chat request with {} messages", msg_count); + + let mut messages = payload.messages; + + // Inject current date awareness if not already present or as a fresh system message + let now = chrono::Local::now(); + let date_str = now.format("%A, %B %e, %Y at %l:%M %P").to_string(); + messages.insert( + 0, + Message { + role: "system".to_string(), + content: Some(format!( + "The current date and time is {}. Today is {}. You are in interactive chat mode.", + date_str, + now.format("%Y-%m-%d") + )), + tool_calls: None, + tool_call_id: None, + }, + ); + + let mut agent = crate::domain::agent::Agent::with_messages( + state.db.clone(), + state.config.zen_api_key.clone(), + state.config.tavily_api_key.clone(), + messages, + )?; + + tracing::info!("Starting interactive agent turn"); + let assistant_message = agent.execute_turn().await?; + + Ok(Json(ChatResult { + message: assistant_message, + })) +} diff --git a/src/server/mod.rs b/src/server/mod.rs index 79d28e2..6979d6e 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,4 +1,6 @@ pub mod auth; +pub mod chat; +pub mod notifications; pub mod tasks; use axum::{ @@ -23,6 +25,7 @@ pub struct AppState { pub config: Arc, pub verifier: Arc, pub authenticator: Arc, + pub tx: tokio::sync::broadcast::Sender, } pub async fn start(config: crate::config::Config) -> AppResult<()> { @@ -30,8 +33,10 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> { let config = Arc::new(config); + let (tx, _) = tokio::sync::broadcast::channel(100); + let scheduler = Arc::new( - Scheduler::new(db.clone(), config.clone()) + Scheduler::new(db.clone(), config.clone(), tx.clone()) .await .map_err(|e| crate::error::AppError::Internal(e.to_string()))?, ); @@ -55,6 +60,7 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> { config: config.clone(), verifier, authenticator, + tx, }); let app = build_app(state, &config); @@ -120,6 +126,8 @@ fn build_app(state: Arc, config: &crate::config::Config) -> Router { .route("/api/auth/callback", get(auth::auth_callback)) .route("/api/auth/refresh", post(auth::auth_refresh)) .route("/api/auth/logout", post(auth::auth_logout)) + .route("/api/chat", post(chat::chat_handler)) + .route("/api/ws", get(notifications::ws_handler)) .layer(cors) .layer(tower_http::set_header::SetResponseHeaderLayer::overriding( axum::http::header::CONTENT_SECURITY_POLICY, diff --git a/src/server/notifications.rs b/src/server/notifications.rs new file mode 100644 index 0000000..dcc62a3 --- /dev/null +++ b/src/server/notifications.rs @@ -0,0 +1,46 @@ +use crate::domain::tasks::{RecentRunResponse, TaskResponse}; +use crate::server::AppState; +use axum::{ + extract::{ + State, + ws::{Message, WebSocket, WebSocketUpgrade}, + }, + response::IntoResponse, +}; +use serde::Serialize; +use std::sync::Arc; + +#[derive(Serialize, Clone, Debug)] +#[serde(tag = "type", content = "data")] +pub enum WsEvent { + TaskCreated(TaskResponse), + TaskUpdated(TaskResponse), + RunStarted(RecentRunResponse), + RunFinished(TaskResponse), +} + +pub async fn ws_handler( + ws: WebSocketUpgrade, + State(state): State>, +) -> impl IntoResponse { + ws.on_upgrade(|socket| handle_socket(socket, state)) +} + +async fn handle_socket(mut socket: WebSocket, state: Arc) { + let mut rx = state.tx.subscribe(); + + while let Ok(event) = rx.recv().await { + let msg = match serde_json::to_string(&event) { + Ok(json) => json, + Err(e) => { + tracing::error!("Failed to serialize WsEvent: {}", e); + continue; + } + }; + + if socket.send(Message::Text(msg)).await.is_err() { + // Client disconnected + break; + } + } +} diff --git a/src/server/tasks.rs b/src/server/tasks.rs index 9782b9c..5c5da2b 100644 --- a/src/server/tasks.rs +++ b/src/server/tasks.rs @@ -84,7 +84,13 @@ pub async fn create_task( let _ = state.scheduler.add_task_job(task_id, cron).await; } - tasks::get_task_inner(task_id, &state.db).await.map(Json) + let task_response = tasks::get_task_inner(task_id, &state.db).await?; + let _ = state + .tx + .send(crate::server::notifications::WsEvent::TaskCreated( + task_response.clone(), + )); + Ok(Json(task_response)) } pub async fn rerun_task( @@ -150,7 +156,13 @@ pub async fn update_task( let _ = state.scheduler.remove_task_job(id).await; } - tasks::get_task_inner(id, &state.db).await.map(Json) + let task_response = tasks::get_task_inner(id, &state.db).await?; + let _ = state + .tx + .send(crate::server::notifications::WsEvent::TaskUpdated( + task_response.clone(), + )); + Ok(Json(task_response)) } pub async fn get_task(