diff --git a/AGENTS.md b/AGENTS.md deleted file mode 100644 index 907bd9b..0000000 --- a/AGENTS.md +++ /dev/null @@ -1,154 +0,0 @@ -# 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 8030dcb..73d8ffb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -161,7 +161,6 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", - "base64", "bytes", "futures-util", "http", @@ -180,10 +179,8 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sha1", "sync_wrapper", "tokio", - "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -339,7 +336,7 @@ dependencies = [ "sea-orm-migration", "serde", "serde_json", - "thiserror 2.0.18", + "thiserror", "tokio", "tokio-cron-scheduler", "tower-http 0.5.2", @@ -670,12 +667,6 @@ 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" @@ -2514,7 +2505,7 @@ dependencies = [ "serde_json", "sqlx", "strum 0.26.3", - "thiserror 2.0.18", + "thiserror", "time", "tracing", "url", @@ -2611,7 +2602,7 @@ dependencies = [ "proc-macro2", "quote", "syn 2.0.114", - "thiserror 2.0.18", + "thiserror", ] [[package]] @@ -2823,7 +2814,7 @@ checksum = "297f631f50729c8c99b84667867963997ec0b50f32b2a7dbcab828ef0541e8bb" dependencies = [ "num-bigint", "num-traits", - "thiserror 2.0.18", + "thiserror", "time", ] @@ -2921,7 +2912,7 @@ dependencies = [ "serde_json", "sha2", "smallvec", - "thiserror 2.0.18", + "thiserror", "time", "tokio", "tokio-stream", @@ -3009,7 +3000,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror 2.0.18", + "thiserror", "time", "tracing", "uuid", @@ -3052,7 +3043,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror 2.0.18", + "thiserror", "time", "tracing", "uuid", @@ -3079,7 +3070,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror 2.0.18", + "thiserror", "time", "tracing", "url", @@ -3230,33 +3221,13 @@ 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 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", + "thiserror-impl", ] [[package]] @@ -3410,18 +3381,6 @@ 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" @@ -3595,24 +3554,6 @@ 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" @@ -3670,12 +3611,6 @@ 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 e7a13dc..6d3bcb2 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 = { version = "0.7", features = ["ws"] } +axum = "0.7" 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 5be239e..ea8c9bd 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -63,56 +63,22 @@
-

Perspective

-

Overview of recent agent directives and live communication.

- -
- - -
+

Recent Activity

+

Track latest agent executions across all directives.

- -
-
-
-
-

Neural Link

- -
-
- -
-

Establish connection with the neural assistant.

-
-
-
- - -
-
-
- -
-
- - - - - - - - - - - -
DirectiveStatusDate
-
-
+
+ + + + + + + + + + + +
DirectiveStatusDate
diff --git a/frontend/src/main.js b/frontend/src/main.js index 91e3ffd..64ef57e 100644 --- a/frontend/src/main.js +++ b/frontend/src/main.js @@ -16,9 +16,7 @@ const state = { selectedRunId: null, currentView: 'dashboard', // 'dashboard' or 'task' isEditing: false, - isAuthenticated: false, - chatMessages: [], - activeDashboardTab: 'chat' // 'chat' or 'activity' + isAuthenticated: false }; // DOM elements const loginOverlay = document.getElementById('login-overlay'); @@ -55,11 +53,6 @@ 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); @@ -96,7 +89,6 @@ function renderApp() { if (state.currentView === 'dashboard') { fetchRecentRuns(); - renderChat(); } else if (state.selectedTaskId) { const task = state.tasks.find(t => t.id === state.selectedTaskId); if (task) { @@ -272,28 +264,6 @@ 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) { @@ -384,62 +354,6 @@ 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; @@ -533,19 +447,6 @@ 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); @@ -610,86 +511,17 @@ async function checkSession() { return false; } -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) => { +let isPolling = false; +async function startAutoRefresh() { + setInterval(async () => { + if (isPolling) return; + isPolling = true; try { - 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); + await fetchTasks(); + } finally { + isPolling = false; } - }; - - 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(); - }; + }, 3000); } async function showLogin() { @@ -746,13 +578,6 @@ 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(); @@ -760,7 +585,7 @@ async function initializeApp() { appEl.classList.remove('hidden'); loginOverlay.classList.add('hidden'); await fetchTasks(); - connectWebSocket(); + startAutoRefresh(); } else { showLogin(); } diff --git a/frontend/src/style.css b/frontend/src/style.css index f34da31..0761f29 100644 --- a/frontend/src/style.css +++ b/frontend/src/style.css @@ -88,15 +88,9 @@ body { .dashboard-content { flex: 1; overflow-y: auto; - border-radius: 16px; + border-radius: 12px; 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 { @@ -130,158 +124,6 @@ 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; } @@ -954,9 +796,9 @@ textarea:focus { } .btn-sm { - padding: 6px 16px; + padding: 6px 12px; font-size: 12px; - width: auto; + width: 100%; justify-content: center; } diff --git a/frontend/vite.config.js b/frontend/vite.config.js index a71c6f4..7ce721b 100644 --- a/frontend/vite.config.js +++ b/frontend/vite.config.js @@ -6,7 +6,6 @@ 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 f886c34..2d2ae37 100644 --- a/src/domain/agent/api.rs +++ b/src/domain/agent/api.rs @@ -3,6 +3,7 @@ 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>, @@ -28,7 +29,6 @@ pub struct FunctionCall { pub struct ChatRequest { pub model: String, pub messages: Vec, - #[serde(skip_serializing_if = "Option::is_none")] pub tools: Option>, } @@ -72,7 +72,6 @@ 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()?; @@ -95,7 +94,6 @@ 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 1b6dde8..9f6933f 100644 --- a/src/domain/agent/mod.rs +++ b/src/domain/agent/mod.rs @@ -2,18 +2,16 @@ 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, - pub messages: Vec, + messages: Vec, tools: Option>, logs: String, answer: Option, @@ -23,7 +21,6 @@ use crate::error::{AppError, AppResult}; impl Agent { pub fn new( - db: DatabaseConnection, zen_api_key: Option, tavily_api_key: Option, initial_message: String, @@ -50,24 +47,14 @@ 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(120)) + .timeout(std::time::Duration::from_secs(60)) .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, @@ -114,35 +101,6 @@ 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 @@ -159,24 +117,20 @@ impl Agent { } } - if let Some(tool_calls) = &assistant_message.tool_calls { - let mut is_final_cycle = false; - let mut final_answer = None; - + if let Some(tool_calls) = assistant_message.tool_calls { 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, &self.db) + tools::handle_tool_call(&tool_call, &self.tavily_api_key) .await .map_err(|e| { AppError::Internal(format!("Tool execution failed: {}", e)) })?; if let Some(ans) = tool_answer { - self.answer = Some(ans.clone()); - final_answer = Some(ans); - self.log("Interaction marked as finished by tool."); + self.answer = Some(ans); + self.log("Task marked as finished by tool."); } if let Some(content) = &tool_message.content { @@ -185,24 +139,20 @@ impl Agent { self.messages.push(tool_message); if is_final { - is_final_cycle = true; + finished = true; } } - - 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; + } 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. } - - return Ok(assistant_message); } + + self.log("\n--- Execution Finished ---"); + Ok((self.logs.clone(), self.answer.clone())) } async fn call_llm(&self) -> AppResult { @@ -218,46 +168,23 @@ impl Agent { request_builder = request_builder.header("Authorization", format!("Bearer {}", key)); } - 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); + let response = request_builder.send().await.map_err(AppError::Network)?; if !response.status().is_success() { let status = response.status(); - let body_text = response + let error_text = response .text() .await - .unwrap_or_else(|_| "Unknown body".into()); - let err = format!("API request failed: {} - {}", status, body_text); - tracing::error!("{}", err); - return Err(AppError::Internal(err)); + .unwrap_or_else(|_| "Unknown error".into()); + return Err(AppError::Internal(format!( + "API request failed: {} - {}", + status, error_text + ))); } - 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) - }) + response + .json() + .await + .map_err(|e| AppError::Internal(format!("Failed to parse LLM response: {}", e))) } } diff --git a/src/domain/agent/tools.rs b/src/domain/agent/tools.rs index bbdff4c..663b1b1 100644 --- a/src/domain/agent/tools.rs +++ b/src/domain/agent/tools.rs @@ -1,9 +1,5 @@ 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 { @@ -15,13 +11,6 @@ struct FinishArgs { result: String, } -#[derive(Deserialize)] -struct ListRunsArgs { - from: Option, - to: Option, - task_ids: Option>, -} - pub fn get_tools() -> Vec { vec![ Tool { @@ -58,51 +47,12 @@ 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; @@ -126,78 +76,6 @@ 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 6e7962d..5092091 100644 --- a/src/domain/tasks.rs +++ b/src/domain/tasks.rs @@ -22,7 +22,7 @@ pub struct UpdateTaskRequest { pub cron: Option, } -#[derive(Serialize, Clone, Debug)] +#[derive(Serialize)] pub struct TaskResponse { pub id: Uuid, pub goal: String, @@ -31,7 +31,7 @@ pub struct TaskResponse { pub runs: Vec, } -#[derive(Serialize, Clone, Debug)] +#[derive(Serialize)] pub struct TaskRunResponse { pub id: Uuid, pub status: String, @@ -40,7 +40,7 @@ pub struct TaskRunResponse { pub created_at: chrono::DateTime, } -#[derive(Serialize, Clone, Debug)] +#[derive(Serialize)] pub struct RecentRunResponse { pub id: Uuid, pub task_id: Uuid, @@ -75,20 +75,7 @@ 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(), @@ -125,14 +112,7 @@ pub async fn execute_agent_run( .await .map_err(crate::error::AppError::Database)?; - 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) + get_task_inner(task_id, db).await } pub async fn get_task_inner(id: Uuid, db: &DatabaseConnection) -> AppResult { diff --git a/src/scheduler.rs b/src/scheduler.rs index 9aad91b..2ee8a01 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -13,14 +13,12 @@ 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 @@ -34,7 +32,6 @@ impl Scheduler { db, tasks_to_jobs: DashMap::new(), config, - tx, }) } @@ -46,14 +43,12 @@ 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, tx, task_id).await { + if let Err(e) = Self::run_task(db, config, task_id).await { tracing::error!("Error in scheduled task {}: {}", task_id, e); } }) @@ -86,7 +81,6 @@ 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) @@ -111,19 +105,8 @@ 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(), @@ -159,12 +142,6 @@ 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 deleted file mode 100644 index 5163275..0000000 --- a/src/server/chat.rs +++ /dev/null @@ -1,57 +0,0 @@ -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 6979d6e..79d28e2 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,6 +1,4 @@ pub mod auth; -pub mod chat; -pub mod notifications; pub mod tasks; use axum::{ @@ -25,7 +23,6 @@ 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<()> { @@ -33,10 +30,8 @@ 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(), tx.clone()) + Scheduler::new(db.clone(), config.clone()) .await .map_err(|e| crate::error::AppError::Internal(e.to_string()))?, ); @@ -60,7 +55,6 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> { config: config.clone(), verifier, authenticator, - tx, }); let app = build_app(state, &config); @@ -126,8 +120,6 @@ 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 deleted file mode 100644 index dcc62a3..0000000 --- a/src/server/notifications.rs +++ /dev/null @@ -1,46 +0,0 @@ -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 5c5da2b..9782b9c 100644 --- a/src/server/tasks.rs +++ b/src/server/tasks.rs @@ -84,13 +84,7 @@ pub async fn create_task( let _ = state.scheduler.add_task_job(task_id, cron).await; } - 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)) + tasks::get_task_inner(task_id, &state.db).await.map(Json) } pub async fn rerun_task( @@ -156,13 +150,7 @@ pub async fn update_task( let _ = state.scheduler.remove_task_job(id).await; } - 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)) + tasks::get_task_inner(id, &state.db).await.map(Json) } pub async fn get_task(