pavel/bot
1
0
Fork 0
Code Issues Pull requests Projects Releases 22 Packages Wiki Activity Actions

Compare commits

base: pavel:main
Branches Tags
pavel:main
pavel:0.0.22
pavel:0.0.21
pavel:0.0.20
pavel:0.0.19
pavel:0.0.18
pavel:0.0.17
pavel:0.0.16
pavel:0.0.15
pavel:0.0.14
pavel:0.0.13
pavel:0.0.12
pavel:0.0.11
pavel:0.0.10
pavel:0.0.9
pavel:0.0.8
pavel:0.0.7
pavel:0.0.6
pavel:0.0.5
pavel:0.0.4
pavel:0.0.3
pavel:0.0.2
pavel:0.0.1
..
compare: pavel:0.0.6
Branches Tags
pavel:main
pavel:0.0.22
pavel:0.0.21
pavel:0.0.20
pavel:0.0.19
pavel:0.0.18
pavel:0.0.17
pavel:0.0.16
pavel:0.0.15
pavel:0.0.14
pavel:0.0.13
pavel:0.0.12
pavel:0.0.11
pavel:0.0.10
pavel:0.0.9
pavel:0.0.8
pavel:0.0.7
pavel:0.0.6
pavel:0.0.5
pavel:0.0.4
pavel:0.0.3
pavel:0.0.2
pavel:0.0.1

No commits in common. "main" and "0.0.6" have entirely different histories.
main ... 0.0.6

36 changed files with 144 additions and 2753 deletions
Download patch file Download diff file Expand all files Collapse all files

25
.forgejo/workflows/pipeline.yaml
View file

@ -11,31 +11,6 @@ jobs:
with:
node-version: 24
- uses: actions/checkout@v6
- name: Cache Node.js modules
uses: actions/cache@v4
with:
path: frontend/node_modules
key: ${{ runner.os }}-node-${{ hashFiles('frontend/package-lock.json') }}
restore-keys: |
${{ runner.os }}-node-
- name: Cache Cargo registry
uses: actions/cache@v4
with:
path: |
~/.cargo/bin/
~/.cargo/registry/index/
~/.cargo/registry/cache/
~/.cargo/git/db/
key: ${{ runner.os }}-cargo-${{ hashFiles('Cargo.lock') }}
restore-keys: |
${{ runner.os }}-cargo-
- name: Cache Cargo target
uses: actions/cache@v4
with:
path: target/
key: ${{ runner.os }}-cargo-target-${{ hashFiles('Cargo.lock') }}
restore-keys: |
${{ runner.os }}-cargo-target-
- run: |
cd frontend
npm install

1
.gitignore vendored
View file

@ -1,4 +1,3 @@
/target
node_modules
/frontend/dist
.env

691
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

2
Cargo.toml
View file

@ -22,8 +22,6 @@ tokio-cron-scheduler = "0.15.1"
dashmap = "6.1.0"
jsonwebtoken = { version = "10.3.0", features = ["rsa", "rust_crypto"] }
base64 = "0.22.1"
web-push = { version = "0.10.0", features = ["isahc-client"] }
isahc = "1.7"
axum-extra = { version = "0.9", features = ["typed-header", "cookie"] }
cookie = "0.18"
thiserror = "2.0.18"

9
frontend/index.html
View file

@ -4,12 +4,8 @@
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0, viewport-fit=cover">
<meta name="theme-color" content="#0a0a0c">
<meta name="apple-mobile-web-app-capable" content="yes">
<meta name="apple-mobile-web-app-status-bar-style" content="black-translucent">
<link rel="apple-touch-icon" href="/apple-touch-icon.png">
<link rel="manifest" href="/manifest.json">
<title>Antigravity Agent Dashboard</title>
<link rel="stylesheet" href="/src/style.css">
<link rel="preconnect" href="https://fonts.googleapis.com">
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
<link
@ -144,9 +140,6 @@
<button id="rerun-btn" class="btn btn-ghost">
<span>↻</span> Re-run Task
</button>
<button id="notify-task-btn" class="btn btn-ghost">
<span class="notify-icon">🔔</span> Notify Me
</button>
</div>
</header>

BIN
frontend/public/apple-touch-icon.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 30 KiB

BIN
frontend/public/icon-192.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 33 KiB

BIN
frontend/public/icon-512.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 27 KiB

21
frontend/public/manifest.json
View file

@ -1,21 +0,0 @@
{
"name": "Antigravity Agency Dashboard",
"short_name": "Agency",
"description": "Autonomous AI Agent Dashboard",
"start_url": "/",
"display": "standalone",
"background_color": "#0a0a0c",
"theme_color": "#5d5dff",
"icons": [
{
"src": "icon-192.png",
"sizes": "192x192",
"type": "image/png"
},
{
"src": "icon-512.png",
"sizes": "512x512",
"type": "image/png"
}
]
}

119
frontend/public/sw.js
View file

@ -1,119 +0,0 @@
const CACHE_NAME = 'agency-cache-v4';
const ASSETS = [
'/',
'/index.html',
'/manifest.json',
'/icon-192.png',
'/icon-512.png'
];
// Force immediate update to the latest SW
self.addEventListener('install', (event) => {
event.waitUntil(
caches.open(CACHE_NAME).then((cache) => {
return cache.addAll(ASSETS);
}).then(() => self.skipWaiting())
);
});
// Clean up old caches and take control of all clients immediately
self.addEventListener('activate', (event) => {
event.waitUntil(
caches.keys().then((cacheNames) => {
return Promise.all(
cacheNames.map((cacheName) => {
if (cacheName !== CACHE_NAME) {
console.log('Deleting old cache:', cacheName);
return caches.delete(cacheName);
}
})
);
}).then(() => self.clients.claim())
);
});
self.addEventListener('fetch', (event) => {
// Only intercept http/https requests
if (!event.request.url.startsWith('http')) return;
event.respondWith(
caches.match(event.request).then((cachedResponse) => {
if (cachedResponse) {
return cachedResponse;
}
return fetch(event.request).catch((error) => {
// If network fetch fails and it's a navigation request, return index.html
if (event.request.mode === 'navigate') {
return caches.match('/index.html');
}
// For assets, return a failure response instead of throwing.
// Re-throwing (or returning a rejected promise) causes the browser to show
// the "unexpected error" interception UI.
console.warn('Fetch failed for:', event.request.url, error);
return new Response('Network error occurred', {
status: 503,
statusText: 'Service Unavailable',
headers: new Headers({ 'Content-Type': 'text/plain' })
});
});
})
);
});
self.addEventListener('push', (event) => {
let data = { title: 'Notification', body: 'New update from Agency' };
try {
if (event.data) {
data = event.data.json();
}
} catch (e) {
console.error('Error parsing push data:', e);
}
const options = {
body: data.body,
icon: '/icon-192.png',
badge: '/icon-192.png',
vibrate: [100, 50, 100],
data: {
dateOfArrival: Date.now(),
primaryKey: '1',
taskId: data.task_id,
runId: data.run_id
}
};
event.waitUntil(
self.registration.showNotification(data.title, options)
);
});
self.addEventListener('notificationclick', (event) => {
event.notification.close();
const taskId = event.notification.data.taskId;
const runId = event.notification.data.runId;
let url = '/';
if (taskId && runId) {
url = `/?taskId=${taskId}&runId=${runId}`;
}
event.waitUntil(
clients.matchAll({ type: 'window', includeUncontrolled: true }).then((windowClients) => {
// Check if there is already a window open and focus it, or open a new one
for (let client of windowClients) {
if ('focus' in client) {
// Navigate the existing client to the new URL if it's the same app
return client.navigate(url).then(c => c.focus());
}
}
if (clients.openWindow) {
return clients.openWindow(url);
}
})
);
});

165
frontend/src/main.js
View file

@ -1,6 +1,5 @@
import { marked } from 'marked';
import DOMPurify from 'dompurify';
import './style.css';
const API_URL = '/api';
// These should ideally be environment-specific
@ -19,8 +18,7 @@ const state = {
isEditing: false,
isAuthenticated: false,
chatMessages: [],
activeDashboardTab: 'chat', // 'chat' or 'activity'
swRegistration: null
activeDashboardTab: 'chat' // 'chat' or 'activity'
};
// DOM elements
const loginOverlay = document.getElementById('login-overlay');
@ -159,16 +157,6 @@ async function fetchTasks() {
const response = await fetchWithAuth(`${API_URL}/tasks`);
const newTasks = await response.json();
// Check for deep link in URL
const params = new URLSearchParams(window.location.search);
const urlTaskId = params.get('taskId');
const urlRunId = params.get('runId');
if (urlTaskId && !state.selectedTaskId) {
state.selectedTaskId = urlTaskId;
state.selectedRunId = urlRunId;
}
// Check if we should follow the latest run
let newSelectedRunId = state.selectedRunId;
if (state.selectedTaskId) {
@ -189,12 +177,6 @@ async function fetchTasks() {
tasks: newTasks,
selectedRunId: newSelectedRunId
});
// If we just loaded from a deep link, clear the params and select it
if (urlTaskId) {
window.history.replaceState({}, document.title, "/");
selectTask(urlTaskId, urlRunId);
}
} catch (error) {
console.error('Error fetching tasks:', error);
}
@ -262,19 +244,6 @@ function selectTask(id, runId = null) {
renderRunHistory(task);
showTaskView(task);
// Fetch and update subscription status
fetch(`${API_URL}/tasks/${id}/subscription`)
.then(res => res.json())
.then(data => {
const btn = document.getElementById('notify-task-btn');
if (data.isSubscribed) {
btn.classList.add('notified');
} else {
btn.classList.remove('notified');
}
})
.catch(err => console.error('Failed to fetch subscription status', err));
// Close sidebar on mobile after selection
if (window.innerWidth <= 768) {
closeMobileMenu();
@ -759,14 +728,7 @@ async function handleCallback() {
const response = await fetch(`${API_URL}/auth/callback?code=${code}&redirect_uri=${encodeURIComponent(AUTH_CONFIG.redirectUri)}`, {
credentials: 'include'
});
const contentType = response.headers.get("content-type");
let data;
if (contentType && contentType.includes("application/json")) {
data = await response.json();
} else {
const text = await response.text();
throw new Error(`Expected JSON but got ${contentType}. Body: ${text.substring(0, 100)}`);
}
const data = await response.json();
if (response.ok) {
state.isAuthenticated = true;
@ -822,16 +784,6 @@ async function initializeApp() {
if (hasSession) {
appEl.classList.remove('hidden');
loginOverlay.classList.add('hidden');
// Handle deep links from notifications
const params = new URLSearchParams(window.location.search);
const taskId = params.get('taskId');
const runId = params.get('runId');
if (taskId) {
state.selectedTaskId = taskId;
state.selectedRunId = runId;
}
await fetchTasks();
connectWebSocket();
} else {
@ -845,116 +797,3 @@ if (window.location.pathname === '/callback' || window.location.search.includes(
} else {
initializeApp();
}
// Register Service Worker for PWA
if ('serviceWorker' in navigator) {
window.addEventListener('load', () => {
navigator.serviceWorker.register('/sw.js')
.then(reg => {
console.log('SW registered', reg);
state.swRegistration = reg;
})
.catch(err => {
console.error('SW registration failed:', err);
if (window.isSecureContext === false) {
console.error('Context is NOT secure. Service Workers require HTTPS or localhost.');
}
});
});
}
async function setupPush() {
if (!state.swRegistration) {
console.warn('SW registration not available');
return false;
}
try {
const vapidResponse = await fetch(`${API_URL}/notifications/vapid-key`);
const { publicKey } = await vapidResponse.json();
// Always clear existing subscription to ensure we use latest VAPID key
const existingSub = await state.swRegistration.pushManager.getSubscription();
if (existingSub) {
await existingSub.unsubscribe();
console.log('Unsubscribed existing push subscription');
}
const subscription = await state.swRegistration.pushManager.subscribe({
userVisibleOnly: true,
applicationServerKey: urlBase64ToUint8Array(publicKey)
});
await fetch(`${API_URL}/notifications/register`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
endpoint: subscription.endpoint,
p256dh: b64(subscription.getKey('p256dh')),
auth: b64(subscription.getKey('auth'))
})
});
console.log('Push registered');
return true;
} catch (err) {
console.warn('Push registration failed:', err);
return false;
}
}
function b64(buffer) {
const binary = String.fromCharCode.apply(null, new Uint8Array(buffer));
return btoa(binary)
.replace(/\+/g, '-')
.replace(/\//g, '_')
.replace(/=/g, '');
}
function urlBase64ToUint8Array(base64String) {
const padding = '='.repeat((4 - base64String.length % 4) % 4);
const base64 = (base64String + padding).replace(/\-/g, '+').replace(/_/g, '/');
const rawData = window.atob(base64);
const outputArray = new Uint8Array(rawData.length);
for (let i = 0; i < rawData.length; ++i) {
outputArray[i] = rawData.charCodeAt(i);
}
return outputArray;
}
async function toggleTaskSubscription(taskId) {
const btn = document.getElementById('notify-task-btn');
const isNotified = btn.classList.contains('notified');
const method = isNotified ? 'DELETE' : 'POST';
// If trying to enable but no push subscription, try setting it up first (user gesture here)
if (!isNotified && 'Notification' in window) {
if (Notification.permission !== 'granted') {
const permission = await Notification.requestPermission();
if (permission !== 'granted') {
showToast('Notification permission denied', 'error');
return;
}
}
const sub = await state.swRegistration.pushManager.getSubscription();
if (!sub) {
const success = await setupPush();
if (!success) {
showToast('Failed to initialize push notifications', 'error');
return;
}
}
}
try {
await fetch(`${API_URL}/tasks/${taskId}/subscribe`, { method });
btn.classList.toggle('notified');
showToast(isNotified ? 'Notifications disabled' : 'Notifications enabled');
} catch (err) {
showToast('Failed to update notifications');
}
}
document.getElementById('notify-task-btn').addEventListener('click', () => {
if (state.selectedTaskId) toggleTaskSubscription(state.selectedTaskId);
});

87
frontend/src/style.css
View file

@ -1101,37 +1101,6 @@ textarea:focus {
}
/* Responsive Styles */
#notify-task-btn.notified {
color: var(--primary);
background: var(--primary-glow);
}
#notify-task-btn.notified .notify-icon {
animation: ring 0.5s ease;
}
@keyframes ring {
0% {
transform: rotate(0);
}
25% {
transform: rotate(15deg);
}
50% {
transform: rotate(-15deg);
}
75% {
transform: rotate(10deg);
}
100% {
transform: rotate(0);
}
}
@media (max-width: 768px) {
.menu-toggle {
display: flex;
@ -1163,9 +1132,6 @@ textarea:focus {
height: auto;
min-height: 100%;
overflow-y: auto;
overflow-x: hidden;
width: 100%;
max-width: 100vw;
}
.dashboard-header h2 {
@ -1176,15 +1142,12 @@ textarea:focus {
flex-direction: column;
height: auto;
overflow: visible;
width: 100%;
gap: 16px;
}
.run-history {
width: 100%;
max-height: 200px;
flex-shrink: 0;
margin-bottom: 8px;
}
.run-details {
@ -1192,14 +1155,6 @@ textarea:focus {
height: auto;
overflow: visible;
gap: 16px;
width: 100%;
}
.answer-container {
width: 100%;
margin-bottom: 8px;
border-left: none;
border-top: 4px solid var(--primary);
}
.logs-container {
@ -1214,27 +1169,14 @@ textarea:focus {
gap: 16px;
}
.header-main {
flex-direction: column;
align-items: flex-start;
gap: 12px;
width: 100%;
}
.header-main h2 {
font-size: 20px;
line-height: 1.3;
}
.header-actions {
width: 100%;
display: flex;
flex-wrap: wrap;
display: grid;
grid-template-columns: 1fr 1fr;
gap: 8px;
}
.header-actions .btn {
flex: 1 1 auto;
justify-content: center;
}
@ -1257,29 +1199,4 @@ textarea:focus {
textarea {
font-size: 16px !important;
}
/* Force word break for long text in containers */
.answer-output,
.activity-table td {
word-break: break-word;
}
.activity-table td {
padding: 12px 10px;
}
/* Hide Date on very small mobile to prevent table overflow */
@media (max-width: 480px) {
.activity-table th:last-child,
.activity-table td:last-child {
display: none;
}
}
/* Prevent horizontal scroll on the entire app */
#app {
width: 100vw;
overflow-x: hidden;
}
}

2
frontend/vite.config.js
View file

@ -4,7 +4,7 @@ export default defineConfig({
server: {
proxy: {
'/api': {
target: 'http://localhost:3001',
target: 'http://localhost:3000',
changeOrigin: true,
ws: true,
}

2
migration/src/lib.rs
View file

@ -4,7 +4,6 @@ mod m20220101_000001_create_table;
mod m20260210_000002_add_answer_column;
mod m20260210_000003_separate_runs;
mod m20260210_000004_add_cron_column;
mod m20260212_000005_notifications;
pub struct Migrator;
@ -16,7 +15,6 @@ impl MigratorTrait for Migrator {
Box::new(m20260210_000002_add_answer_column::Migration),
Box::new(m20260210_000003_separate_runs::Migration),
Box::new(m20260210_000004_add_cron_column::Migration),
Box::new(m20260212_000005_notifications::Migration),
]
}
}

115
migration/src/m20260212_000005_notifications.rs
View file

@ -1,115 +0,0 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Push Subscriptions table
manager
.create_table(
Table::create()
.table(PushSubscriptions::Table)
.if_not_exists()
.col(
ColumnDef::new(PushSubscriptions::Id)
.uuid()
.not_null()
.primary_key(),
)
.col(
ColumnDef::new(PushSubscriptions::UserSub)
.string()
.not_null(),
)
.col(
ColumnDef::new(PushSubscriptions::Endpoint)
.string()
.not_null(),
)
.col(
ColumnDef::new(PushSubscriptions::P256dh)
.string()
.not_null(),
)
.col(ColumnDef::new(PushSubscriptions::Auth).string().not_null())
.col(
ColumnDef::new(PushSubscriptions::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
// Task Subscriptions table
manager
.create_table(
Table::create()
.table(TaskSubscriptions::Table)
.if_not_exists()
.col(
ColumnDef::new(TaskSubscriptions::Id)
.uuid()
.not_null()
.primary_key(),
)
.col(
ColumnDef::new(TaskSubscriptions::UserSub)
.string()
.not_null(),
)
.col(ColumnDef::new(TaskSubscriptions::TaskId).uuid().not_null())
.col(
ColumnDef::new(TaskSubscriptions::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
ForeignKey::create()
.name("fk-task-subscription-task-id")
.from(TaskSubscriptions::Table, TaskSubscriptions::TaskId)
.to(Tasks::Table, Tasks::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.drop_table(Table::drop().table(TaskSubscriptions::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(PushSubscriptions::Table).to_owned())
.await
}
}
#[derive(DeriveIden)]
enum PushSubscriptions {
Table,
Id,
UserSub,
Endpoint,
P256dh,
Auth,
CreatedAt,
}
#[derive(DeriveIden)]
enum TaskSubscriptions {
Table,
Id,
UserSub,
TaskId,
CreatedAt,
}
#[derive(DeriveIden)]
enum Tasks {
Table,
Id,
}

593
openapi/calendar/openapi.json
View file

@ -1,593 +0,0 @@
{
"openapi": "3.1.0",
"info": {
"title": "calendar",
"description": "",
"license": {
"name": ""
},
"version": "0.1.0"
},
"paths": {
"/auth/me": {
"get": {
"tags": [
"crate::handlers::auth"
],
"operationId": "me",
"responses": {
"200": {
"description": "Current user profile",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/CurrentUser"
}
}
}
},
"401": {
"description": "Unauthorized"
}
},
"security": [
{
"oidc": []
}
]
}
},
"/events": {
"get": {
"tags": [
"crate::handlers::event"
],
"operationId": "list_events",
"parameters": [
{
"name": "upcoming",
"in": "query",
"required": false,
"schema": {
"type": [
"boolean",
"null"
]
}
}
],
"responses": {
"200": {
"description": "List of events",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Model"
}
}
}
}
},
"401": {
"description": "Unauthorized"
}
},
"security": [
{
"oidc": []
}
]
},
"post": {
"tags": [
"crate::handlers::event"
],
"operationId": "create_event",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/CreateEventRequest"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Event created successfully",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"400": {
"description": "Invalid request payload"
},
"401": {
"description": "Unauthorized"
}
},
"security": [
{
"oidc": []
}
]
}
},
"/events/{id}": {
"get": {
"tags": [
"crate::handlers::event"
],
"operationId": "get_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"responses": {
"200": {
"description": "Event details",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
},
"put": {
"tags": [
"crate::handlers::event"
],
"operationId": "update_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/CreateEventRequest"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Event updated successfully",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"400": {
"description": "Invalid request payload"
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
},
"delete": {
"tags": [
"crate::handlers::event"
],
"operationId": "delete_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"responses": {
"204": {
"description": "Event deleted successfully"
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
}
},
"/service/v1/events": {
"get": {
"tags": [
"crate::handlers::service"
],
"operationId": "service_list_events",
"parameters": [
{
"name": "user_id",
"in": "query",
"required": false,
"schema": {
"type": [
"integer",
"null"
],
"format": "int32"
}
},
{
"name": "upcoming",
"in": "query",
"required": false,
"schema": {
"type": [
"boolean",
"null"
]
}
}
],
"responses": {
"200": {
"description": "List of events",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Model"
}
}
}
}
},
"401": {
"description": "Unauthorized"
}
},
"security": [
{
"oidc": []
}
]
},
"post": {
"tags": [
"crate::handlers::service"
],
"operationId": "service_create_event",
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceCreateEventRequest"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Event created successfully",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"400": {
"description": "Invalid request payload"
},
"401": {
"description": "Unauthorized"
}
},
"security": [
{
"oidc": []
}
]
}
},
"/service/v1/events/{id}": {
"get": {
"tags": [
"crate::handlers::service"
],
"operationId": "service_get_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"responses": {
"200": {
"description": "Event details",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
},
"put": {
"tags": [
"crate::handlers::service"
],
"operationId": "service_update_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ServiceCreateEventRequest"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Event updated successfully",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Model"
}
}
}
},
"400": {
"description": "Invalid request payload"
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
},
"delete": {
"tags": [
"crate::handlers::service"
],
"operationId": "service_delete_event",
"parameters": [
{
"name": "id",
"in": "path",
"description": "Event database id",
"required": true,
"schema": {
"type": "integer",
"format": "int32"
}
}
],
"responses": {
"204": {
"description": "Event deleted successfully"
},
"401": {
"description": "Unauthorized"
},
"404": {
"description": "Event not found"
}
},
"security": [
{
"oidc": []
}
]
}
}
},
"components": {
"schemas": {
"CreateEventRequest": {
"type": "object",
"required": [
"name",
"from",
"to"
],
"properties": {
"from": {
"type": "string"
},
"name": {
"type": "string"
},
"to": {
"type": "string"
}
}
},
"CurrentUser": {
"type": "object",
"required": [
"id",
"sub",
"email",
"name"
],
"properties": {
"email": {
"type": "string"
},
"id": {
"type": "integer",
"format": "int32"
},
"name": {
"type": "string"
},
"sub": {
"type": "string"
}
}
},
"Model": {
"type": "object",
"required": [
"id",
"name",
"from",
"to"
],
"properties": {
"from": {
"type": "string",
"format": "date-time"
},
"id": {
"type": "integer",
"format": "int64"
},
"name": {
"type": "string"
},
"to": {
"type": "string",
"format": "date-time"
},
"user_id": {
"type": [
"integer",
"null"
],
"format": "int32"
}
}
},
"ServiceCreateEventRequest": {
"type": "object",
"required": [
"name",
"from",
"to"
],
"properties": {
"from": {
"type": "string"
},
"name": {
"type": "string"
},
"to": {
"type": "string"
},
"user_id": {
"type": [
"integer",
"null"
],
"format": "int32"
}
}
}
}
},
"tags": [
{
"name": "calendar",
"description": "Calendar Management API"
}
]
}

10
src/config.rs
View file

@ -14,8 +14,6 @@ pub struct Config {
pub cookie_secure: bool,
pub agent_max_turns: u32,
pub agent_max_duration_secs: u64,
pub vapid_private_key: String,
pub calendar_api_url: String,
}
impl Config {
@ -56,12 +54,6 @@ impl Config {
.and_then(|s| s.parse().ok())
.unwrap_or(120);
let vapid_private_key = env::var("VAPID_PRIVATE_KEY")
.map_err(|_| AppError::Config("VAPID_PRIVATE_KEY must be set".into()))?;
let calendar_api_url =
env::var("CALENDAR_API_URL").unwrap_or_else(|_| "http://localhost:8000".to_string());
Ok(Config {
database_url,
port,
@ -74,8 +66,6 @@ impl Config {
cookie_secure,
agent_max_turns,
agent_max_duration_secs,
vapid_private_key,
calendar_api_url,
})
}
}

2
src/domain/agent/api.rs
View file

@ -75,8 +75,6 @@ pub async fn perform_search(
tracing::info!(query = %query, "Performing Tavily web search");
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(30))
.connect_timeout(std::time::Duration::from_secs(10))
.pool_idle_timeout(std::time::Duration::from_secs(60))
.build()?;
let response = client
.post("https://api.tavily.com/search")

79
src/domain/agent/mod.rs
View file

@ -3,7 +3,6 @@ pub mod tools;
use chrono::Utc;
use sea_orm::DatabaseConnection;
use std::sync::Arc;
use std::time::{Duration, Instant};
use self::api::{ChatRequest, ChatResponse, Message, Tool};
@ -14,8 +13,6 @@ pub struct Agent {
url: String,
zen_api_key: Option<String>,
tavily_api_key: Option<String>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
pub user_sub: Option<String>,
pub messages: Vec<Message>,
tools: Option<Vec<Tool>>,
logs: String,
@ -29,8 +26,6 @@ impl Agent {
db: DatabaseConnection,
zen_api_key: Option<String>,
tavily_api_key: Option<String>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
user_sub: Option<String>,
initial_message: String,
) -> AppResult<Self> {
let intro = format!(
@ -55,31 +50,19 @@ impl Agent {
},
];
Self::with_messages(
db,
zen_api_key,
tavily_api_key,
calendar_client,
user_sub,
messages,
)
Self::with_messages(db, zen_api_key, tavily_api_key, messages)
}
pub fn with_messages(
db: DatabaseConnection,
zen_api_key: Option<String>,
tavily_api_key: Option<String>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
user_sub: Option<String>,
messages: Vec<Message>,
) -> AppResult<Self> {
let tools = Some(tools::get_tools());
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(120))
.connect_timeout(std::time::Duration::from_secs(10))
.tcp_keepalive(std::time::Duration::from_secs(30))
.pool_idle_timeout(std::time::Duration::from_secs(60))
.build()
.map_err(|e| AppError::Internal(format!("Failed to build HTTP client: {}", e)))?;
@ -89,8 +72,6 @@ impl Agent {
url: "https://opencode.ai/zen/v1/chat/completions".to_string(),
zen_api_key,
tavily_api_key,
calendar_client,
user_sub,
messages,
tools,
logs: String::new(),
@ -122,8 +103,6 @@ impl Agent {
return Err(AppError::Internal("Agent run exceeded max turns".into()));
}
tracing::info!("Turn {}", turns);
turns += 1;
let current_role = self
.messages
@ -135,39 +114,17 @@ impl Agent {
turns, current_role
));
let assistant_message =
match tokio::time::timeout(Duration::from_secs(180), self.execute_turn()).await {
Ok(res) => res?,
Err(_) => {
tracing::error!("Agent execution turn timed out after 180s");
return Err(AppError::Internal("Agent execution turn timed out".into()));
}
};
let assistant_message = self.execute_turn().await?;
if let Some(tool_calls) = &assistant_message.tool_calls {
tracing::info!("Assistant tool calls: {:#?}", tool_calls);
if tool_calls.iter().any(|tc| tc.function.name == "answer") {
finished = true;
}
}
if self.answer.is_some() {
tracing::info!("Answer: {}", self.answer.as_ref().unwrap());
finished = true;
}
if !finished {
self.messages.push(Message {
role: "system".to_string(),
content: Some(
"continue, use the finish tool to submit your final answer".to_string(),
),
tool_calls: None,
tool_call_id: None,
});
}
tracing::info!("Finished turn");
}
self.log("\n--- Execution Finished ---");
@ -175,12 +132,10 @@ impl Agent {
}
pub async fn execute_turn(&mut self) -> AppResult<Message> {
let max_sub_turns = 20;
let max_sub_turns = 10;
let mut sub_turns = 0;
loop {
tracing::info!("Sub turn {}", sub_turns);
sub_turns += 1;
if sub_turns > max_sub_turns {
return Err(AppError::Internal(
@ -196,34 +151,27 @@ impl Agent {
.message
.clone();
tracing::info!("Assistant message: {:#?}", assistant_message);
self.messages.push(assistant_message.clone());
if let Some(content) = &assistant_message.content {
tracing::info!("Assistant content: {}", content);
if !content.is_empty() {
self.log(&format!("\nAssistant: {}", content));
}
}
if let Some(tool_calls) = &assistant_message.tool_calls {
tracing::info!("Assistant tool calls: {:#?}", 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,
&self.db,
&self.calendar_client,
self.user_sub.as_deref(),
)
let (tool_message, is_final, tool_answer) =
tools::handle_tool_call(tool_call, &self.tavily_api_key, &self.db)
.await
.map_err(|e| AppError::Internal(format!("Tool execution failed: {}", e)))?;
.map_err(|e| {
AppError::Internal(format!("Tool execution failed: {}", e))
})?;
if let Some(ans) = tool_answer {
self.answer = Some(ans.clone());
@ -242,7 +190,6 @@ impl Agent {
}
if is_final_cycle {
tracing::info!("Final answer: {}", final_answer.as_ref().unwrap());
return Ok(Message {
role: "assistant".to_string(),
content: final_answer.or(assistant_message.content),
@ -265,11 +212,7 @@ impl Agent {
tools: self.tools.clone(),
};
let mut request_builder = self
.client
.post(&self.url)
.json(&request)
.timeout(Duration::from_secs(60));
let mut request_builder = self.client.post(&self.url).json(&request);
if let Some(key) = &self.zen_api_key {
request_builder = request_builder.header("Authorization", format!("Bearer {}", key));
@ -279,12 +222,10 @@ impl Agent {
let response = request_builder.send().await.map_err(|e| {
let duration = start.elapsed();
let is_timeout = e.is_timeout();
let is_connect = e.is_connect();
tracing::error!(
"Network error after {:?} during LLM call (Timeout: {}, Connect: {}): {:?}",
"Network error after {:?} during LLM call (Timeout: {}): {:?}",
duration,
is_timeout,
is_connect,
e
);
AppError::Network(e)

72
src/domain/agent/tools.rs
View file

@ -3,7 +3,6 @@ use sea_orm::{
ColumnTrait, Condition, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder, QuerySelect,
};
use serde::Deserialize;
use std::sync::Arc;
use uuid::Uuid;
#[derive(Deserialize)]
@ -97,47 +96,6 @@ pub fn get_tools() -> Vec<Tool> {
}),
},
},
Tool {
tool_type: "function".to_string(),
function: FunctionDefinition {
name: "calendar_list_events".to_string(),
description: "List calendar events".to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"upcoming": {
"type": "boolean",
"description": "If true, only upcoming events will be listed"
}
}
}),
},
},
Tool {
tool_type: "function".to_string(),
function: FunctionDefinition {
name: "calendar_create_event".to_string(),
description: "Create a new calendar event".to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name of the event"
},
"from": {
"type": "string",
"description": "Start time in ISO 8601 format (e.g., 2023-10-27T10:00:00Z)"
},
"to": {
"type": "string",
"description": "End time in ISO 8601 format (e.g., 2023-10-27T11:00:00Z)"
}
},
"required": ["name", "from", "to"]
}),
},
},
]
}
@ -145,8 +103,6 @@ pub async fn handle_tool_call(
tool_call: &ToolCall,
tavily_api_key: &Option<String>,
db: &DatabaseConnection,
calendar: &Arc<crate::domain::calendar::CalendarClient>,
user_sub: Option<&str>,
) -> Result<(Message, bool, Option<String>), Box<dyn std::error::Error>> {
let mut answer = None;
let name = &tool_call.function.name;
@ -242,34 +198,6 @@ pub async fn handle_tool_call(
));
}
(out, false)
} else if name == "calendar_list_events" {
let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)?;
let upcoming = args["upcoming"].as_bool();
match calendar
.list_events(user_sub.map(|s| s.to_string()), upcoming)
.await
{
Ok(events) => {
tracing::info!("{:#?}", events);
(serde_json::to_string(&events)?, false)
}
Err(e) => (format!("Error listing events: {}", e), false),
}
} else if name == "calendar_create_event" {
let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)?;
let name_val = args["name"].as_str().unwrap_or_default();
let from_val = args["from"].as_str().unwrap_or_default();
let to_val = args["to"].as_str().unwrap_or_default();
match calendar
.create_event(user_sub.map(|s| s.to_string()), name_val, from_val, to_val)
.await
{
Ok(event) => (
format!("Event created: {}", serde_json::to_string(&event)?),
false,
),
Err(e) => (format!("Error creating event: {}", e), false),
}
} else {
(format!("Error: Unknown tool {}", name), false)
};

33
src/domain/auth.rs
View file

@ -46,10 +46,7 @@ pub struct JwksVerifier {
impl JwksVerifier {
pub async fn new(issuer: String, audience: String) -> Result<Self, Box<dyn std::error::Error>> {
let client = Client::builder()
.timeout(std::time::Duration::from_secs(30))
.connect_timeout(std::time::Duration::from_secs(10))
.build()?;
let client = Client::new();
let discovery_url = format!(
"{}/.well-known/openid-configuration",
issuer.trim_end_matches('/')
@ -126,10 +123,7 @@ impl Authenticator {
client_id: String,
client_secret: String,
) -> Result<Self, Box<dyn std::error::Error>> {
let client = Client::builder()
.timeout(std::time::Duration::from_secs(30))
.connect_timeout(std::time::Duration::from_secs(10))
.build()?;
let client = Client::new();
let discovery_url = format!(
"{}/.well-known/openid-configuration",
issuer.trim_end_matches('/')
@ -196,27 +190,4 @@ impl Authenticator {
Ok(res)
}
pub async fn client_credentials(
&self,
scope: &str,
) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
let params = [
("grant_type", "client_credentials"),
("client_id", &self.client_id),
("client_secret", &self.client_secret),
("scope", scope),
];
let res = self
.client
.post(&self.token_url)
.form(&params)
.send()
.await?
.json()
.await?;
Ok(res)
}
}

298
src/domain/calendar/mod.rs
View file

@ -1,298 +0,0 @@
use crate::domain::auth::Authenticator;
use crate::error::{AppError, AppResult};
use chrono::{DateTime, Duration, Utc};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::RwLock;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CalendarEvent {
pub id: Option<i64>,
pub name: String,
pub from: DateTime<Utc>,
pub to: DateTime<Utc>,
pub user_sub: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CreateEventRequest {
pub name: String,
pub from: String,
pub to: String,
pub user_sub: Option<String>,
}
struct TokenState {
access_token: String,
expires_at: DateTime<Utc>,
}
pub struct CalendarClient {
base_url: String,
client: Client,
authenticator: Arc<Authenticator>,
token_state: RwLock<Option<TokenState>>,
}
impl CalendarClient {
pub fn new(base_url: String, authenticator: Arc<Authenticator>) -> Self {
let client = Client::builder()
.timeout(std::time::Duration::from_secs(30))
.connect_timeout(std::time::Duration::from_secs(10))
.build()
.unwrap_or_else(|_| Client::new());
Self {
base_url: base_url.trim_end_matches('/').to_string(),
client,
authenticator,
token_state: RwLock::new(None),
}
}
async fn get_token(&self) -> AppResult<String> {
{
let state = self.token_state.read().await;
if let Some(token) = &*state {
if token.expires_at > Utc::now() + Duration::seconds(30) {
tracing::debug!("Using cached Calendar API token");
return Ok(token.access_token.clone());
}
}
}
let mut state = self.token_state.write().await;
// Double check after acquiring write lock
if let Some(token) = &*state {
if token.expires_at > Utc::now() + Duration::seconds(30) {
return Ok(token.access_token.clone());
}
}
tracing::info!("Refreshing Calendar API token via Client Credentials flow");
let token_data = self
.authenticator
.client_credentials("profile")
.await
.map_err(|e| AppError::Internal(format!("Failed to get client credentials: {}", e)))?;
let access_token = token_data["access_token"]
.as_str()
.ok_or_else(|| AppError::Internal("Missing access_token in response".into()))?
.to_string();
let expires_in = token_data["expires_in"].as_i64().unwrap_or(3600);
let expires_at = Utc::now() + Duration::seconds(expires_in);
*state = Some(TokenState {
access_token: access_token.clone(),
expires_at,
});
Ok(access_token)
}
pub async fn list_events(
&self,
user_sub: Option<String>,
upcoming: Option<bool>,
) -> AppResult<Vec<CalendarEvent>> {
let token = self.get_token().await?;
let mut url = format!("{}/service/v1/events", self.base_url);
let mut params = Vec::new();
if let Some(uid) = &user_sub {
params.push(format!("user_sub={}", uid));
}
if let Some(u) = upcoming {
params.push(format!("upcoming={}", u));
}
if !params.is_empty() {
url.push_str("?");
url.push_str(&params.join("&"));
}
tracing::info!(method = "GET", %url, "Sending Calendar API request");
let res = self
.client
.get(&url)
.bearer_auth(token)
.send()
.await
.map_err(AppError::Network)?;
let status = res.status();
tracing::info!(%status, %url, "Received Calendar API response");
if !status.is_success() {
let error_body = res.text().await.unwrap_or_default();
tracing::error!(%status, %url, body = %error_body, "Calendar API request failed");
return Err(AppError::Internal(format!(
"Failed to list events: {} - {}",
status, error_body
)));
}
let body = res
.json()
.await
.map_err(|e| AppError::Internal(e.to_string()));
tracing::info!(%status, %url, body = ?body, "Received Calendar API response");
body
}
pub async fn create_event(
&self,
user_sub: Option<String>,
name: &str,
from: &str,
to: &str,
) -> AppResult<CalendarEvent> {
let token = self.get_token().await?;
let url = format!("{}/service/v1/events", self.base_url);
let request = CreateEventRequest {
name: name.to_string(),
from: from.to_string(),
to: to.to_string(),
user_sub,
};
tracing::info!(method = "POST", %url, "Sending Calendar API request");
let res = self
.client
.post(&url)
.bearer_auth(token)
.json(&request)
.send()
.await
.map_err(AppError::Network)?;
let status = res.status();
tracing::info!(%status, %url, "Received Calendar API response");
if !status.is_success() {
let error_body = res.text().await.unwrap_or_default();
tracing::error!(%status, %url, body = %error_body, "Calendar API request failed");
return Err(AppError::Internal(format!(
"Failed to create event: {} - {}",
status, error_body
)));
}
res.json()
.await
.map_err(|e| AppError::Internal(e.to_string()))
}
#[allow(dead_code)]
pub async fn get_event(&self, id: i32) -> AppResult<CalendarEvent> {
let token = self.get_token().await?;
let url = format!("{}/service/v1/events/{}", self.base_url, id);
tracing::info!(method = "GET", %url, "Sending Calendar API request");
let res = self
.client
.get(&url)
.bearer_auth(token)
.send()
.await
.map_err(AppError::Network)?;
let status = res.status();
tracing::info!(%status, %url, "Received Calendar API response");
if !status.is_success() {
let error_body = res.text().await.unwrap_or_default();
tracing::error!(%status, %url, body = %error_body, "Calendar API request failed");
return Err(AppError::Internal(format!(
"Failed to get event: {} - {}",
status, error_body
)));
}
res.json()
.await
.map_err(|e| AppError::Internal(e.to_string()))
}
#[allow(dead_code)]
pub async fn update_event(
&self,
id: i32,
user_sub: Option<String>,
name: &str,
from: &str,
to: &str,
) -> AppResult<CalendarEvent> {
let token = self.get_token().await?;
let url = format!("{}/service/v1/events/{}", self.base_url, id);
let request = CreateEventRequest {
name: name.to_string(),
from: from.to_string(),
to: to.to_string(),
user_sub,
};
tracing::info!(method = "PUT", %url, "Sending Calendar API request");
let res = self
.client
.put(&url)
.bearer_auth(token)
.json(&request)
.send()
.await
.map_err(AppError::Network)?;
let status = res.status();
tracing::info!(%status, %url, "Received Calendar API response");
if !status.is_success() {
let error_body = res.text().await.unwrap_or_default();
tracing::error!(%status, %url, body = %error_body, "Calendar API request failed");
return Err(AppError::Internal(format!(
"Failed to update event: {} - {}",
status, error_body
)));
}
res.json()
.await
.map_err(|e| AppError::Internal(e.to_string()))
}
#[allow(dead_code)]
pub async fn delete_event(&self, id: i32) -> AppResult<()> {
let token = self.get_token().await?;
let url = format!("{}/service/v1/events/{}", self.base_url, id);
tracing::info!(method = "DELETE", %url, "Sending Calendar API request");
let res = self
.client
.delete(&url)
.bearer_auth(token)
.send()
.await
.map_err(AppError::Network)?;
let status = res.status();
tracing::info!(%status, %url, "Received Calendar API response");
if !status.is_success() {
let error_body = res.text().await.unwrap_or_default();
tracing::error!(%status, %url, body = %error_body, "Calendar API request failed");
return Err(AppError::Internal(format!(
"Failed to delete event: {} - {}",
status, error_body
)));
}
Ok(())
}
}

2
src/domain/mod.rs
View file

@ -1,5 +1,3 @@
pub mod agent;
pub mod auth;
pub mod calendar;
pub mod notifications;
pub mod tasks;

1
src/domain/notifications/mod.rs
View file

@ -1 +0,0 @@
pub mod push;

113
src/domain/notifications/push.rs
View file

@ -1,113 +0,0 @@
use crate::error::AppResult;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use web_push::*;
pub struct PushSender {
pub private_key: String,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct PushSubscription {
pub endpoint: String,
pub p256dh: String,
pub auth: String,
}
fn ensure_pem(input: &str) -> String {
let input = input.trim();
if input.contains("-----BEGIN") {
return input.to_string();
}
if input.starts_with("MHc") {
format!(
"-----BEGIN EC PRIVATE KEY-----\n{}\n-----END EC PRIVATE KEY-----",
input
)
} else {
format!(
"-----BEGIN PRIVATE KEY-----\n{}\n-----END PRIVATE KEY-----",
input
)
}
}
impl PushSender {
pub fn new(private_key_pem: &str) -> AppResult<Self> {
let pem = ensure_pem(private_key_pem);
// Validate key immediately to catch config errors early
let _ =
VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&pem)).map_err(|e| {
crate::error::AppError::Internal(format!("Invalid VAPID private key: {}", e))
})?;
Ok(Self { private_key: pem })
}
pub fn get_public_key(&self) -> AppResult<Vec<u8>> {
let builder =
VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&self.private_key))
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
Ok(builder.get_public_key())
}
pub async fn send_notification(
&self,
subscription: &PushSubscription,
title: &str,
body: &str,
task_id: Option<Uuid>,
run_id: Option<Uuid>,
) -> AppResult<()> {
let subscription_info = SubscriptionInfo::new(
subscription.endpoint.clone(),
subscription.p256dh.clone(),
subscription.auth.clone(),
);
let builder =
VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&self.private_key))
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
let mut builder = builder.add_sub_info(&subscription_info);
builder.add_claim("sub", "mailto:pavel@flegr.me");
let vapid_signature = builder
.build()
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
let mut builder = WebPushMessageBuilder::new(&subscription_info);
builder.set_vapid_signature(vapid_signature);
let payload = serde_json::to_vec(&serde_json::json!({
"title": title,
"body": body,
"task_id": task_id,
"run_id": run_id,
}))
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
builder.set_payload(ContentEncoding::Aes128Gcm, &payload);
let message = builder
.build()
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
let client = IsahcWebPushClient::new()
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
client.send(message).await.map_err(|e| {
tracing::error!("Failed to send push notification: {}", e);
crate::error::AppError::Internal(e.to_string())
})?;
tracing::info!(
"Push notification sent successfully to {}",
subscription.endpoint
);
Ok(())
}
}

90
src/domain/tasks.rs
View file

@ -1,5 +1,5 @@
use chrono::Utc;
use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};
use sea_orm::{ActiveModelTrait, DatabaseConnection, EntityTrait, Set};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use uuid::Uuid;
@ -8,7 +8,6 @@ use crate::config::Config;
use crate::domain::agent::Agent;
use crate::entities::task::Entity as Task;
use crate::entities::task_run::{self, Entity as TaskRun};
use crate::entities::{push_subscription, task_subscription};
use crate::scheduler::Scheduler;
#[derive(Deserialize)]
@ -56,12 +55,11 @@ pub async fn execute_agent_run(
db: &DatabaseConnection,
_scheduler: &Arc<Scheduler>,
config: &Arc<Config>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
task_id: Uuid,
goal: String,
) -> AppResult<TaskResponse> {
let run_id = Uuid::new_v4();
tracing::info!(%task_id, %run_id, "Starting background agent execution run");
tracing::info!(%task_id, %run_id, "Starting agent execution run");
let new_run = task_run::ActiveModel {
id: Set(run_id),
@ -93,24 +91,16 @@ pub async fn execute_agent_run(
db.clone(),
config.zen_api_key.clone(),
config.tavily_api_key.clone(),
calendar_client.clone(),
None,
goal.clone(),
)?;
let db_bg = db.clone();
let config_bg = config.clone();
let scheduler_bg = _scheduler.clone();
let task_id_bg = task_id;
tokio::spawn(async move {
let (logs, answer, status) = match agent.run(&config_bg).await {
let (logs, answer, status) = match agent.run(config).await {
Ok((logs, answer)) => {
tracing::info!(task_id = %task_id_bg, run_id = %run_id, "Agent execution completed successfully");
tracing::info!(%task_id, %run_id, "Agent execution completed successfully");
(logs, answer, "completed".to_string())
}
Err(e) => {
tracing::error!(task_id = %task_id_bg, run_id = %run_id, error = %e, "Agent execution failed");
tracing::error!(%task_id, %run_id, error = %e, "Agent execution failed");
(
format!("Execution failed: {}", e),
None,
@ -119,66 +109,30 @@ pub async fn execute_agent_run(
}
};
let run_update = task_run::ActiveModel {
id: Set(run_id),
logs: Set(logs),
answer: Set(answer),
status: Set(status.clone()),
..Default::default()
};
let run: task_run::ActiveModel = TaskRun::find_by_id(run_id)
.one(db)
.await
.map_err(crate::error::AppError::Database)?
.ok_or_else(|| crate::error::AppError::NotFound("Run not found after insert".into()))?
.into();
if let Err(e) = run_update.update(&db_bg).await {
tracing::error!(task_id = %task_id_bg, run_id = %run_id, error = %e, "Failed to update run record");
}
let mut run = run;
run.logs = Set(logs.clone());
run.answer = Set(answer.clone());
run.status = Set(status);
if let Ok(task_response) = get_task_inner(task_id_bg, &db_bg).await {
let _ = scheduler_bg
run.update(db)
.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(),
));
// Send Push Notifications to subscribers
if let Ok(subscriptions) = task_subscription::Entity::find()
.filter(task_subscription::Column::TaskId.eq(task_id_bg))
.all(&db_bg)
.await
{
for sub in subscriptions {
if let Ok(push_subs) = push_subscription::Entity::find()
.filter(push_subscription::Column::UserSub.eq(sub.user_sub.clone()))
.all(&db_bg)
.await
{
for push_sub in push_subs {
let sender = scheduler_bg.push_sender.clone();
let goal = task_response.goal.clone();
let status_bg = status.clone();
let sub_data = crate::domain::notifications::push::PushSubscription {
endpoint: push_sub.endpoint,
p256dh: push_sub.p256dh,
auth: push_sub.auth,
};
tokio::spawn(async move {
let _ = sender
.send_notification(
&sub_data,
&format!("Task Completed: {}", status_bg),
&goal,
Some(task_id_bg),
Some(run_id),
)
.await;
});
}
}
}
}
}
});
get_task_inner(task_id, db).await
Ok(task_response)
}
pub async fn get_task_inner(id: Uuid, db: &DatabaseConnection) -> AppResult<TaskResponse> {

2
src/entities/mod.rs
View file

@ -1,4 +1,2 @@
pub mod push_subscription;
pub mod task;
pub mod task_run;
pub mod task_subscription;

19
src/entities/push_subscription.rs
View file

@ -1,19 +0,0 @@
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
#[sea_orm(table_name = "push_subscriptions")]
pub struct Model {
#[sea_orm(primary_key, auto_increment = false)]
pub id: Uuid,
pub user_sub: String,
pub endpoint: String,
pub p256dh: String,
pub auth: String,
pub created_at: DateTimeWithTimeZone,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {}
impl ActiveModelBehavior for ActiveModel {}

32
src/entities/task_subscription.rs
View file

@ -1,32 +0,0 @@
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
#[sea_orm(table_name = "task_subscriptions")]
pub struct Model {
#[sea_orm(primary_key, auto_increment = false)]
pub id: Uuid,
pub user_sub: String,
pub task_id: Uuid,
pub created_at: DateTimeWithTimeZone,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {
#[sea_orm(
belongs_to = "super::task::Entity",
from = "Column::TaskId",
to = "super::task::Column::Id",
on_update = "NoAction",
on_delete = "Cascade"
)]
Task,
}
impl Related<super::task::Entity> for Entity {
fn to() -> RelationDef {
Relation::Task.def()
}
}
impl ActiveModelBehavior for ActiveModel {}

16
src/error.rs
View file

@ -32,20 +32,16 @@ pub enum AppError {
impl IntoResponse for AppError {
fn into_response(self) -> Response {
let (status, error_message) = match &self {
let (status, error_message) = match self {
AppError::Database(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()),
AppError::Config(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.clone()),
AppError::NotFound(err) => (StatusCode::NOT_FOUND, err.clone()),
AppError::Unauthorized(err) => (StatusCode::UNAUTHORIZED, err.clone()),
AppError::Internal(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.clone()),
AppError::Config(err) => (StatusCode::INTERNAL_SERVER_ERROR, err),
AppError::NotFound(err) => (StatusCode::NOT_FOUND, err),
AppError::Unauthorized(err) => (StatusCode::UNAUTHORIZED, err),
AppError::Internal(err) => (StatusCode::INTERNAL_SERVER_ERROR, err),
AppError::Network(err) => (StatusCode::BAD_GATEWAY, err.to_string()),
AppError::InvalidRequest(err) => (StatusCode::BAD_REQUEST, err.clone()),
AppError::InvalidRequest(err) => (StatusCode::BAD_REQUEST, err),
};
if status.is_server_error() || status.is_client_error() {
tracing::error!(%status, error = %self, "AppError converted to response");
}
let body = Json(json!({
"error": error_message,
}));

62
src/scheduler.rs
View file

@ -13,18 +13,14 @@ pub struct Scheduler {
db: DatabaseConnection,
tasks_to_jobs: DashMap<Uuid, Uuid>,
config: Arc<crate::config::Config>,
pub calendar_client: Arc<crate::domain::calendar::CalendarClient>,
pub tx: tokio::sync::broadcast::Sender<crate::server::notifications::WsEvent>,
pub push_sender: Arc<crate::domain::notifications::push::PushSender>,
}
impl Scheduler {
pub async fn new(
db: DatabaseConnection,
config: Arc<crate::config::Config>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
tx: tokio::sync::broadcast::Sender<crate::server::notifications::WsEvent>,
push_sender: Arc<crate::domain::notifications::push::PushSender>,
) -> AppResult<Self> {
let scheduler = JobScheduler::new()
.await
@ -38,9 +34,7 @@ impl Scheduler {
db,
tasks_to_jobs: DashMap::new(),
config,
calendar_client,
tx,
push_sender,
})
}
@ -53,19 +47,13 @@ impl Scheduler {
let db = self.db.clone();
let config = self.config.clone();
let tx = self.tx.clone();
let push_sender = self.push_sender.clone();
let calendar_client = self.calendar_client.clone();
let job = Job::new_async(cron_expr, move |_uuid, _l| {
let db = db.clone();
let config = config.clone();
let tx = tx.clone();
let push_sender = push_sender.clone();
let calendar_client = calendar_client.clone();
Box::pin(async move {
if let Err(e) =
Self::run_task(db, config, calendar_client, tx, push_sender, 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);
}
})
@ -98,9 +86,7 @@ impl Scheduler {
async fn run_task(
db: DatabaseConnection,
config: Arc<crate::config::Config>,
calendar_client: Arc<crate::domain::calendar::CalendarClient>,
tx: tokio::sync::broadcast::Sender<crate::server::notifications::WsEvent>,
push_sender: Arc<crate::domain::notifications::push::PushSender>,
task_id: Uuid,
) -> AppResult<()> {
let task = Task::find_by_id(task_id)
@ -140,8 +126,6 @@ impl Scheduler {
db.clone(),
config.zen_api_key.clone(),
config.tavily_api_key.clone(),
calendar_client.clone(),
None,
task.goal.clone(),
)?;
@ -163,7 +147,7 @@ impl Scheduler {
let run_complete = task_run::ActiveModel {
id: Set(run_id),
status: Set(status.clone()),
status: Set(status),
logs: Set(logs),
answer: Set(answer),
..Default::default()
@ -178,48 +162,8 @@ impl Scheduler {
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.clone(),
task_response,
));
// Push notifications for scheduled runs
use crate::entities::{push_subscription, task_subscription};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
if let Ok(subscriptions) = task_subscription::Entity::find()
.filter(task_subscription::Column::TaskId.eq(task_id))
.all(&db)
.await
{
for sub in subscriptions {
if let Ok(push_subs) = push_subscription::Entity::find()
.filter(push_subscription::Column::UserSub.eq(sub.user_sub))
.all(&db)
.await
{
for push_sub in push_subs {
let sender = push_sender.clone();
let sub_data =
crate::domain::notifications::push::PushSubscription {
endpoint: push_sub.endpoint,
p256dh: push_sub.p256dh,
auth: push_sub.auth,
};
let goal = task_response.goal.clone();
let status = status.clone();
tokio::spawn(async move {
let _ = sender
.send_notification(
&sub_data,
&format!("Scheduled Task Completed: {}", status),
&goal,
Some(task_id),
Some(run_id),
)
.await;
});
}
}
}
}
}
});

13
src/server/chat.rs
View file

@ -17,7 +17,6 @@ pub struct ChatResult {
pub async fn chat_handler(
State(state): State<Arc<AppState>>,
user: crate::server::auth::AuthenticatedUser,
Json(payload): Json<ChatPayload>,
) -> Result<Json<ChatResult>, AppError> {
let msg_count = payload.messages.len();
@ -46,21 +45,11 @@ pub async fn chat_handler(
state.db.clone(),
state.config.zen_api_key.clone(),
state.config.tavily_api_key.clone(),
state.calendar_client.clone(),
Some(user.0.sub),
messages,
)?;
tracing::info!("Starting interactive agent turn");
let assistant_message =
match tokio::time::timeout(std::time::Duration::from_secs(180), agent.execute_turn()).await
{
Ok(res) => res?,
Err(_) => {
tracing::error!("Interactive chat agent turn timed out after 180s");
return Err(AppError::Internal("Agent turn timed out".into()));
}
};
let assistant_message = agent.execute_turn().await?;
Ok(Json(ChatResult {
message: assistant_message,

45
src/server/mod.rs
View file

@ -13,6 +13,7 @@ use sea_orm::{Database, DatabaseConnection, EntityTrait};
use std::sync::Arc;
use tower_http::cors::{AllowOrigin, CorsLayer};
use crate::entities::task::Entity as Task;
use crate::scheduler::Scheduler;
use crate::error::AppResult;
@ -24,7 +25,6 @@ pub struct AppState {
pub config: Arc<crate::config::Config>,
pub verifier: Arc<crate::domain::auth::JwksVerifier>,
pub authenticator: Arc<crate::domain::auth::Authenticator>,
pub calendar_client: Arc<crate::domain::calendar::CalendarClient>,
pub tx: tokio::sync::broadcast::Sender<crate::server::notifications::WsEvent>,
}
@ -35,30 +35,13 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> {
let (tx, _) = tokio::sync::broadcast::channel(100);
let push_sender = Arc::new(crate::domain::notifications::push::PushSender::new(
&config.vapid_private_key.clone(),
)?);
let (verifier, authenticator) = setup_auth(&config).await?;
let calendar_client = Arc::new(crate::domain::calendar::CalendarClient::new(
config.calendar_api_url.clone(),
authenticator.clone(),
));
let scheduler = Arc::new(
Scheduler::new(
db.clone(),
config.clone(),
calendar_client.clone(),
tx.clone(),
push_sender.clone(),
)
Scheduler::new(db.clone(), config.clone(), tx.clone())
.await
.map_err(|e| crate::error::AppError::Internal(e.to_string()))?,
);
// Load existing scheduled tasks
use crate::entities::task::Entity as Task;
let existing_tasks = Task::find()
.all(&db)
.await
@ -69,13 +52,14 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> {
}
}
let (verifier, authenticator) = setup_auth(&config).await?;
let state = Arc::new(AppState {
db,
scheduler,
config: config.clone(),
verifier,
authenticator,
calendar_client,
tx,
});
@ -144,11 +128,6 @@ fn build_app(state: Arc<AppState>, config: &crate::config::Config) -> Router {
.route("/api/auth/logout", post(auth::auth_logout))
.route("/api/chat", post(chat::chat_handler))
.route("/api/ws", get(notifications::ws_handler))
.route("/api/notifications/register", post(notifications::push_handlers::register_push))
.route("/api/notifications/vapid-key", get(notifications::push_handlers::get_vapid_key))
.route("/api/tasks/:id/subscription", get(notifications::push_handlers::get_subscription_status))
.route("/api/tasks/:id/subscribe", post(notifications::push_handlers::subscribe_task).delete(notifications::push_handlers::unsubscribe_task))
.layer(axum::middleware::from_fn(log_error_responses))
.layer(cors)
.layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
axum::http::header::CONTENT_SECURITY_POLICY,
@ -166,22 +145,6 @@ fn build_app(state: Arc<AppState>, config: &crate::config::Config) -> Router {
.with_state(state)
}
async fn log_error_responses(
req: axum::extract::Request,
next: axum::middleware::Next,
) -> axum::response::Response {
let method = req.method().clone();
let uri = req.uri().clone();
let res = next.run(req).await;
let status = res.status();
if status.is_client_error() || status.is_server_error() {
tracing::error!(%method, %uri, %status, "Response error");
}
res
}
fn build_cors_layer(config: &crate::config::Config) -> CorsLayer {
let allow_origin = if let Some(origins) = &config.cors_allowed_origins {
let values: Vec<HeaderValue> = origins

2
src/server/notifications.rs
View file

@ -1,5 +1,3 @@
pub mod push_handlers;
use crate::domain::tasks::{RecentRunResponse, TaskResponse};
use crate::server::AppState;
use axum::{

133
src/server/notifications/push_handlers.rs
View file

@ -1,133 +0,0 @@
use axum::{
Json,
extract::{Path, State},
};
use chrono::Utc;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use std::sync::Arc;
use uuid::Uuid;
use crate::entities::{push_subscription, task_subscription};
use crate::error::{AppError, AppResult};
use crate::server::AppState;
use crate::server::auth::AuthenticatedUser;
use base64::Engine;
#[derive(serde::Deserialize)]
pub struct RegisterPushRequest {
pub endpoint: String,
pub p256dh: String,
pub auth: String,
}
pub async fn register_push(
State(state): State<Arc<AppState>>,
user: AuthenticatedUser,
Json(payload): Json<RegisterPushRequest>,
) -> AppResult<Json<serde_json::Value>> {
let user_sub = user.0.sub;
tracing::info!(
"Registering push subscription for user: {} with endpoint: {}",
user_sub,
payload.endpoint
);
// Check if subscription exists
let existing = push_subscription::Entity::find()
.filter(push_subscription::Column::UserSub.eq(user_sub.clone()))
.filter(push_subscription::Column::Endpoint.eq(payload.endpoint.clone()))
.one(&state.db)
.await
.map_err(AppError::Database)?;
if existing.is_none() {
let new_sub = push_subscription::ActiveModel {
id: Set(Uuid::new_v4()),
user_sub: Set(user_sub),
endpoint: Set(payload.endpoint),
p256dh: Set(payload.p256dh),
auth: Set(payload.auth),
created_at: Set(Utc::now().into()),
};
new_sub
.insert(&state.db)
.await
.map_err(AppError::Database)?;
}
Ok(Json(serde_json::json!({ "status": "registered" })))
}
pub async fn subscribe_task(
State(state): State<Arc<AppState>>,
user: AuthenticatedUser,
Path(task_id): Path<Uuid>,
) -> AppResult<Json<serde_json::Value>> {
let user_sub = user.0.sub;
let existing = task_subscription::Entity::find()
.filter(task_subscription::Column::UserSub.eq(user_sub.clone()))
.filter(task_subscription::Column::TaskId.eq(task_id))
.one(&state.db)
.await
.map_err(AppError::Database)?;
if existing.is_none() {
let new_sub = task_subscription::ActiveModel {
id: Set(Uuid::new_v4()),
user_sub: Set(user_sub),
task_id: Set(task_id),
created_at: Set(Utc::now().into()),
};
new_sub
.insert(&state.db)
.await
.map_err(AppError::Database)?;
}
Ok(Json(serde_json::json!({ "status": "subscribed" })))
}
pub async fn unsubscribe_task(
State(state): State<Arc<AppState>>,
user: AuthenticatedUser,
Path(task_id): Path<Uuid>,
) -> AppResult<Json<serde_json::Value>> {
let user_sub = user.0.sub;
task_subscription::Entity::delete_many()
.filter(task_subscription::Column::UserSub.eq(user_sub))
.filter(task_subscription::Column::TaskId.eq(task_id))
.exec(&state.db)
.await
.map_err(AppError::Database)?;
Ok(Json(serde_json::json!({ "status": "unsubscribed" })))
}
pub async fn get_vapid_key(
State(state): State<Arc<AppState>>,
) -> AppResult<Json<serde_json::Value>> {
let public_key = state.scheduler.push_sender.get_public_key()?;
let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(public_key);
Ok(Json(serde_json::json!({ "publicKey": encoded })))
}
pub async fn get_subscription_status(
State(state): State<Arc<AppState>>,
user: AuthenticatedUser,
Path(task_id): Path<Uuid>,
) -> AppResult<Json<serde_json::Value>> {
let user_sub = user.0.sub;
let existing = task_subscription::Entity::find()
.filter(task_subscription::Column::UserSub.eq(user_sub))
.filter(task_subscription::Column::TaskId.eq(task_id))
.one(&state.db)
.await
.map_err(AppError::Database)?;
Ok(Json(
serde_json::json!({ "isSubscribed": existing.is_some() }),
))
}

1
src/server/tasks.rs
View file

@ -109,7 +109,6 @@ pub async fn rerun_task(
&state.db,
&state.scheduler,
&state.config,
state.calendar_client.clone(),
task.id,
task.goal,
)