# Project export: Kettle AI
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2025
- Tagline: Ready to go.
- Devpost: https://devpost.com/software/kettle-ai
- GitHub: https://github.com/rishabh-athreya/Kettle
- Video: https://www.youtube.com/embed/dcGMzJzqbxI?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Rishabh Athreya (2 commits)
## Devpost submission (written by the team)
### Inspiration
When our supervisors send us tasks, a lot of time is spent figuring out how to get started: what to do research on, how to organize the project, and what checklist of items need to be completed to accomplish the objectives. The advent of background agents encouraged us to explore novel applications where the monotonous task of setup and planning for the day would be done for you—before you even open your computer.
### What it does
Kettle AI autonomously monitors and fetches slack messages from your manager, generating a list of tasks categorized into multiple types: research (opening videos, articles, and papers), project setup (folder creation), dependency installation, and implementation (as much code that can be automated for the task). When you open your computer, simply go on the Kettle web app to check all the tasks and setup that we've planned in the background. We give ultimate development power to the user, letting them choose to approve or reject different tasks, with approved tasks being assigned to Kettle AI to complete for you. Whether you want an AI assistant to complete your tasks for you or to just need a boilerplate setup and some guidance on what to do, you choose how Kettle helps you. Just open your computer and start building - no friction, hassle, or setup required!
### How we built it
We used the Slack API to periodically fetch messages from a specified channel. These messages are processed and fed into the Claude API, which generates a task list in a JSON file—annotated with the categories from above. Our Flask backend reads the task list and presents them in an intuitive UI to the React frontend. Once the user checks the tasks they want to see executed, they simply select the preprepared tasks and hit execute, which then sends back the selected tasks to the Kettle backend accomplishing several different functionalities, including either returning resources for research (such as papers, articles, blogs, and videos) with Gemini and Perplexity or creating an executable script with Claude to create folders, install dependencies, and write code (complete with a venv with necessary installations).
### Challenges we ran into
One main challenge we ran into was making sure the user wasn't making any mistakes when choosing their desired tasks. For example, if they had rejected Task A but accepted Task B, and Task B depended on Task A, then this would increase the risk of faulty outputs. Therefore, we prompted Claude again as an intermediary step to generate a "dependency matrix," which tells us which tasks are dependent on each other and when to warn users if a task should probably not be rejected.
### Accomplishments we're proud of
We are most proud of the effective workflow we've designed, which involves autonomous Slack message monitoring, generating useful tasks, allowing the user to have control over selecting the most appropriate preprepared tasks, and execute them directly on your system. While there are many avenues of growth for our existing system, we believe we've taken a massive step forward in helping streamline the productivity process with AI while keeping humans in ultimate control of how they want to build.
### What we learned
We learned how involved prompt engineering could be, and how specific details could change the quality of the output dramatically. We also learned it was important to reason from the standpoint of the user when making design decisions, such as in deciding what components of the workflow the user should have control over and which ones should be completely left to our system.
### What's next
We aim to improve the capabilities of Kettle, including more complex coding tasks, more generalized agentic functionality for report creation and calendar event creation, and integration with Gmail, iMessage, Microsoft Teams, and other communications platforms.
## README (from the GitHub repository)
No README available.
## Detected evidence (automated analysis)
Indexed codebase: 32 recognized source files, 206 KB.
- Anthropic (technology) — detected in the code
- CSS (language) — detected in the code
- Flask (technology) — detected in the code
- Hugging Face (technology) — detected in the code
- JavaScript (language) — detected in the code
- Next.js (technology) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — detected in the code
- React (technology) — detected in the code
- Tailwind CSS (technology) — detected in the code
- TypeScript (language) — detected in the code
- Google Gemini (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (49 of 49)
```
.gitignore
json/coding_tasks.json
json/dependency_matrix.json
json/last_processed_ts.txt
json/last_task_processed_ts.txt
json/media.json
json/messages.json
json/phased_tasks.json
json/project_embeddings.json
json/research_tasks.json
json/task_dependencies.json
json/writing_tasks.json
kettle_monitor.py
kettle.sh
requirements.txt
slack_fetch.py
start_all.sh
tools/__init__.py
tools/check_mongodb_status.py
tools/dependency_analyzer.py
tools/desktop_widget.py
tools/execute_tasks.py
tools/extract_tasks.py
tools/migrate_to_mongodb.py
tools/mongodb_config.py
tools/project_matcher.py
tools/test_mongodb_integration.py
utils/clear_json.py
utils/json_utils.py
utils/prompts.py
web_app/api/app.py
web_app/api/requirements.txt
web_app/app/components/MessageCard.tsx
web_app/app/components/ResearchResourceCard.tsx
web_app/app/components/StatsCard.tsx
web_app/app/components/TaskCard.tsx
web_app/app/globals.css
web_app/app/layout.tsx
web_app/app/page.tsx
web_app/app/types.ts
web_app/next-env.d.ts
web_app/next.config.js
web_app/package.json
web_app/postcss.config.js
web_app/README.md
web_app/setup.sh
web_app/tailwind.config.js
web_app/tsconfig.json
web_app/vercel.json
```
### Dependencies
- requirements.txt: anthropic@==0.34.0, blinker@==1.9.0, certifi@==2025.6.15, charset-normalizer@==3.4.2, click@==8.2.1, filelock@==3.18.0, Flask@==3.1.1, Flask-CORS@==5.0.0, fsspec@==2025.5.1, google-generativeai@==0.8.3, hf-xet@==1.1.4, huggingface-hub@==0.33.0, idna@==3.10, itsdangerous@==2.2.0, Jinja2@==3.1.6, joblib@==1.5.1, MarkupSafe@==3.0.2, mpmath@==1.3.0, networkx@==3.5, numpy@==2.3.0, packaging@==25.0, pillow@==11.2.1, psutil@==7.0.0, pygame@==2.6.1, pymongo@==4.6.1, PyYAML@==6.0.2, regex@==2024.11.6, requests@==2.32.4, safetensors@==0.5.3, scikit-learn@==1.7.0, scipy@==1.15.3, sentence-transformers@==4.1.0, setuptools@==80.9.0, slack_sdk@==3.35.0, SQLAlchemy@==2.0.41, sympy@==1.14.0, threadpoolctl@==3.6.0, tokenizers@==0.21.1, torch@==2.7.1, tqdm@==4.67.1, transformers@==4.52.4, typing_extensions@==4.14.0, urllib3@==2.4.0, Werkzeug@==3.1.3
- web_app/api/requirements.txt: Flask@==3.1.1, Flask-CORS@==4.0.0, Werkzeug@==3.1.3
- web_app/package.json: @types/node@^20, @types/react@^18, @types/react-dom@^18, autoprefixer@^10.0.1, clsx@^2.0.0, date-fns@^2.30.0, eslint@^8, eslint-config-next@14.0.0, framer-motion@^10.16.4, lucide-react@^0.294.0, next@14.0.0, postcss@^8, react@^18, react-dom@^18, tailwindcss@^3.3.0, typescript@^5
### Recent commits (newest first)
- Restructured
- Merged & MongoDB fixes
- MongoDB fixes
- MongoDB integration, file organization
- Intermediate
- Restored thank lord
## Key source files (fetched from GitHub, selected and truncated for size)
### requirements.txt
```
anthropic==0.34.0
blinker==1.9.0
certifi==2025.6.15
charset-normalizer==3.4.2
click==8.2.1
filelock==3.18.0
Flask==3.1.1
Flask-CORS==5.0.0
fsspec==2025.5.1
google-generativeai==0.8.3
hf-xet==1.1.4
huggingface-hub==0.33.0
idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.6
joblib==1.5.1
MarkupSafe==3.0.2
mpmath==1.3.0
networkx==3.5
numpy==2.3.0
packaging==25.0
pillow==11.2.1
psutil==7.0.0
pygame==2.6.1
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.4
safetensors==0.5.3
scikit-learn==1.7.0
scipy==1.15.3
sentence-transformers==4.1.0
setuptools==80.9.0
slack_sdk==3.35.0
SQLAlchemy==2.0.41
sympy==1.14.0
threadpoolctl==3.6.0
tokenizers==0.21.1
torch==2.7.1
tqdm==4.67.1
transformers==4.52.4
typing_extensions==4.14.0
urllib3==2.4.0
Werkzeug==3.1.3
pymongo==4.6.1
```
### web_app/package.json
```
{
"name": "kettle-dashboard",
"version": "0.1.0",
"private": true,
"scripts": {
"dev": "next dev",
"build": "next build",
"start": "next start",
"lint": "next lint"
},
"dependencies": {
"next": "14.0.0",
"react": "^18",
"react-dom": "^18",
"lucide-react": "^0.294.0",
"tailwindcss": "^3.3.0",
"autoprefixer": "^10.0.1",
"postcss": "^8",
"framer-motion": "^10.16.4",
"date-fns": "^2.30.0",
"clsx": "^2.0.0"
},
"devDependencies": {
"typescript": "^5",
"@types/node": "^20",
"@types/react": "^18",
"@types/react-dom": "^18",
"eslint": "^8",
"eslint-config-next": "14.0.0"
}
}
```
### web_app/api/requirements.txt
```
Flask==3.1.1
Flask-CORS==4.0.0
Werkzeug==3.1.3
```
### web_app/app/layout.tsx
```typescript
import React from 'react'
import type { Metadata } from 'next'
import { Inter } from 'next/font/google'
import './globals.css'
const inter = Inter({ subsets: ['latin'] })
export const metadata: Metadata = {
title: 'Kettle AI Dashboard',
description: 'AI-Powered Project Automation Dashboard',
}
export default function RootLayout({
children,
}: {
children: React.ReactNode
}) {
return (
{children}
)
}
```
### web_app/api/app.py
```python
from flask import Flask, jsonify, request, send_from_directory
from flask_cors import CORS
import json
import os
import uuid
import subprocess
import sys
from datetime import datetime
from typing import List, Dict, Any
app = Flask(__name__)
CORS(app)
# Path to the JSON data files (relative to the main Kettle directory)
KETTLE_DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'json')
KETTLE_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
# Path to the virtual environment Python interpreter
VENV_PYTHON = os.path.join(KETTLE_ROOT, 'kettle_env', 'bin', 'python3')
def load_json_file(filename: str) -> Dict[str, Any]:
"""Load JSON file from the Kettle data directory"""
filepath = os.path.join(KETTLE_DATA_DIR, filename)
try:
with open(filepath, 'r') as f:
return json.load(f)
except FileNotFoundError:
return {}
except json.JSONDecodeError:
return {}
def save_json_file(filename: str, data: Dict[str, Any]) -> None:
"""Save JSON file to the Kettle data directory"""
filepath = os.path.join(KETTLE_DATA_DIR, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
def transform_task(task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Transform task data to include required fields for the frontend"""
# Generate a consistent ID based on task content if not present
if 'id' not in task_data:
task_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{task_data.get('task', '')}{task_data.get('source', '')}"))
else:
task_id = task_data['id']
return {
'id': task_id,
'task': task_data.get('task', ''),
'source': task_data.get('source', ''),
'phase': task_data.get('phase', 'feature_implementation'),
'selectionStatus': task_data.get('selectionStatus', 'pending'), # Changed from approvalStatus
'createdAt': task_data.get('createdAt', datetime.now().isoformat()),
'selectedAt': task_data.get('selectedAt'),
'user': task_data.get('user')
}
def all_coding_tasks_executed(tasks_data):
"""Return True if all coding tasks are executed, False otherwise."""
if not isinstance(tasks_data, list):
return False
coding_tasks = [task for task in tasks_data if task.get('category', 'coding') == 'coding']
return all(task.get('selectionStatus') == 'executed' for task in coding_tasks) if coding_tasks else False
@app.route('/api/tasks', methods=['GET'])
def get_tasks():
"""Get coding tasks first; only return research tasks if all coding tasks are executed."""
try:
tasks_data = load_json_file('phased_tasks.json')
if not isinstance(tasks_data, list):
return jsonify([])
if not all_coding_tasks_executed(tasks_data):
# Return only coding tasks
tasks = [transform_task(task) for task in tasks_data if task.get('category', 'coding') == 'coding']
else:
# Return only research tasks
tasks = [transform_task(task) for task in tasks_data if task.get('category') == 'research']
return jsonify(tasks)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/tasks//select', methods=['POST'])
def select_task(task_id: str):
"""Update task selection status for coding tasks"""
try:
data = request.get_json()
status = data.get('status', 'pending') # 'selected', 'rejected', or 'pending'
# Load current tasks
tasks_data = load_json_file('phased_tasks.json')
if isinstance(tasks_data, list):
# Find and update the task (only coding tasks)
task_found = False
for task in tasks_data:
# Only process coding tasks
task_category = task.get('category', 'coding')
if task_category != 'coding':
continue
# Check if this task matches the ID (either by existing ID or by generating one)
task_check_id = task.get('id') or str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{task.get('task', '')}{task.get('source', '')}"))
if task_check_id == task_id:
task['id'] = task_id # Ensure ID is set
task['selectionStatus'] = status
if status in ['selected', 'rejected']:
task['selectedAt'] = datetime.now().isoformat()
task_found = True
break
if not task_found:
return jsonify({'error': 'Coding task not found'}), 404
# Save updated tasks
save_json_file('phased_tasks.json', tasks_data)
return jsonify({'success': True, 'status': status})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/tasks//dependencies', methods=['GET'])
def get_task_dependencies(task_id: str):
"""Get potential dependencies for a task using AI-generated dependency matrix"""
try:
# Load current tasks
tasks_data = load_json_file('phased_tasks.json')
if not isinstance(tasks_data, list):
return jsonify({'error': 'No tasks found'}), 404
# Find the target task
target_task = None
for task in tasks_data:
task_check_id = task.get('id') or str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{task.get('task', '')}{task.get('source', '')}"))
if task_check_id == task_id:
target_task = task
break
if not target_task:
return jsonify({'error': 'Task not found'}), 404
# Load dependency matrix
try:
dependency_matrix = load_json_file('dependency_matrix.json')
dependencies_data
[truncated — 10325 more characters]
```
### web_app/app/page.tsx
```typescript
'use client'
import React, { useState, useEffect, useCallback } from 'react'
import { motion } from 'framer-motion'
import {
Coffee,
CheckCircle,
XCircle,
Clock,
MessageSquare,
Zap,
RefreshCw,
TrendingUp,
Play,
Trash2
} from 'lucide-react'
import TaskCard from './components/TaskCard'
import MessageCard from './components/MessageCard'
import StatsCard from './components/StatsCard'
import ResearchResourceCard, { ResearchResource } from './components/ResearchResourceCard'
import { Task, Message, SelectionStatus } from './types'
export default function Dashboard() {
const [tasks, setTasks] = useState([])
const [messages, setMessages] = useState([])
const [loading, setLoading] = useState(true)
const [executing, setExecuting] = useState(false)
const [lastUpdate, setLastUpdate] = useState(new Date())
const [stats, setStats] = useState({
totalTasks: 0,
selectedTasks: 0,
pendingSelection: 0,
executedTasks: 0,
totalMessages: 0
})
const [resources, setResources] = useState([])
const [writingTasks, setWritingTasks] = useState([])
useEffect(() => {
fetchData()
fetchResearchResources()
fetchWritingTasks()
// Set up polling for real-time updates - reduced to 10 seconds
const interval = setInterval(() => {
console.log('Auto-refreshing data...')
fetchData()
fetchResearchResources()
fetchWritingTasks()
}, 10000)
return () => clearInterval(interval)
}, [])
const fetchData = async () => {
try {
console.log('Fetching data from API...')
const [tasksRes, messagesRes] = await Promise.all([
fetch('/api/tasks'),
fetch('/api/messages')
])
if (tasksRes.ok) {
const tasksData = await tasksRes.json()
console.log(`Fetched ${tasksData.length} tasks`)
setTasks(tasksData)
setStats(prev => ({
...prev,
totalTasks: tasksData.length,
selectedTasks: tasksData.filter((t: Task) => t.selectionStatus === 'selected').length,
pendingSelection: tasksData.filter((t: Task) => t.selectionStatus === 'pending').length,
executedTasks: tasksData.filter((t: Task) => t.selectionStatus === 'executed').length
}))
} else {
console.error('Failed to fetch tasks:', tasksRes.status)
}
if (messagesRes.ok) {
const messagesData = await messagesRes.json()
console.log(`Fetched ${messagesData.messages?.length || 0} messages`)
setMessages(messagesData.messages || [])
setStats(prev => ({
...prev,
totalMessages: messagesData.messages?.length || 0
}))
} else {
console.error('Failed to fetch messages:', messagesRes.status)
}
setLastUpdate(new Date())
} catch (error) {
console.error('Error fetching data:', error)
} finally {
setLoading(false)
}
}
const fetchResearchResources = async () => {
try {
const res = await fetch('/api/research-resources')
if (res.ok) {
const data = await res.json()
setResources(data)
}
} catch (error) {
console.error('Error fetching research resources:', error)
}
}
const fetchWritingTasks = async () => {
try {
const res = await fetch('/api/writing_tasks')
if (res.ok) {
const data = await res.json()
setWritingTasks(data)
}
} catch (error) {
console.error('Error fetching writing tasks:', error)
}
}
const handleTaskSelection = async (taskId: string, status: SelectionStatus) => {
try {
const response = await fetch(`/api/tasks/${taskId}/select`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ status })
})
if (response.ok) {
// Always re-fetch from backend to ensure UI is in sync
await fetchData()
} else {
console.error('Failed to update task selection')
alert('Failed to update task selection. Please try again.')
}
} catch (error) {
console.error('Error updating task selection:', error)
alert('Error updating task selection. Please try again.')
}
}
const handleExecuteSelected = async () => {
const selectedTasks = tasks.filter(task => task.selectionStatus === 'selected')
if (selectedTasks.length === 0) {
alert('No tasks selected for execution. Please select some tasks first.')
return
}
setExecuting(true)
try {
const response = await fetch('/api/execute-selected', {
method: 'POST',
headers: { 'Content-Type': 'application/json' }
})
const result = await response.json()
if (result.success) {
alert(`Successfully executed ${result.executedTasks} tasks!`)
await fetchData() // Refresh to show updated statuses
} else {
console.error('Execution failed:', result.error)
alert(`Execution failed: ${result.error}`)
}
} catch (error) {
console.error('Error executing tasks:', error)
alert('Error executing tasks. Check console for details.')
} finally {
setExecuting(false)
}
}
const handleReset = async () => {
if (!confirm('Are you sure you want to reset all tasks and messages? This will clear everything.')) {
return
}
try {
// Clear tasks and messages by setting them to empty
const tasksResponse = await fetch('/api/tasks', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ action: 'reset' })
})
const messagesResponse = await fetch('/api/messages', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ action: 'reset' })
})
const mediaResponse = await fetch('/api/medi
[truncated — 11382 more characters]
```
### start_all.sh
```shell
#!/bin/bash
# Start Continuous Slack Monitoring (waits until you exit)
echo "==============================="
echo "Starting: Slack Monitoring"
echo "==============================="
source kettle_env/bin/activate
python kettle_monitor.py
echo "==============================="
echo "Starting: Flask API"
echo "==============================="
cd web_app/api
../../kettle_env/bin/python app.py
cd ../..
echo "==============================="
echo "Starting: Next.js Frontend"
echo "==============================="
cd web_app
npm run dev
cd ..
echo "All services have finished running."
```
### kettle.sh
```shell
#!/bin/bash
# Kettle AI Daemon Script
# This script runs multiple services for the Kettle AI system
echo "🚀 Starting Kettle AI Services..."
# Function to kill processes using specific ports
kill_ports() {
PORTS=(3000 5000 7860)
for PORT in "${PORTS[@]}"; do
PIDS=$(lsof -ti tcp:$PORT)
if [ ! -z "$PIDS" ]; then
echo "🔪 Killing processes on port $PORT: $PIDS"
kill -9 $PIDS 2>/dev/null
fi
done
}
# Kill any processes using needed ports before starting
kill_ports
# Function to kill background processes on exit
cleanup() {
echo "🛑 Shutting down Kettle AI services..."
kill $FLASK_PID $MONITOR_PID $NEXTJS_PID 2>/dev/null
exit 0
}
# Set up signal handlers
trap cleanup SIGINT SIGTERM
# Start Flask API (background)
echo "📡 Starting Flask API..."
cd web_app/api
python app.py &
FLASK_PID=$!
cd ../..
# Start Kettle Monitor (background)
echo "👁️ Starting Kettle Monitor..."
python kettle_monitor.py &
MONITOR_PID=$!
# Start Next.js Frontend (background)
echo "🌐 Starting Next.js Frontend..."
cd web_app
npm run dev &
NEXTJS_PID=$!
cd ..
echo "✅ All services started!"
echo "📊 Dashboard: http://localhost:3000"
echo "🔌 API: http://localhost:5000"
echo ""
echo "Press Ctrl+C to stop all services"
# Wait for all background processes
wait
```
### slack_fetch.py
```python
import os
import json
import time
import requests
from datetime import datetime, timezone, timedelta
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import subprocess
import sys
# You can set these in your environment or in utils/keys.py
try:
from utils.keys import SLACK_BOT_TOKEN, SLACK_CHANNEL_ID, BOT_USER_ID
except Exception:
SLACK_BOT_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
SLACK_CHANNEL_ID = os.environ.get("SLACK_CHANNEL_ID")
BOT_USER_ID = os.environ.get("BOT_USER_ID")
assert SLACK_BOT_TOKEN, "SLACK_BOT_TOKEN must be set in environment or keys.py"
assert SLACK_CHANNEL_ID, "SLACK_CHANNEL_ID must be set in environment or keys.py"
client = WebClient(token=SLACK_BOT_TOKEN)
# How many messages to fetch (adjust as needed)
MESSAGE_LIMIT = 50
# Option 2: Fetch bot user ID from Slack if not set
if not BOT_USER_ID:
auth_resp = client.auth_test()
BOT_USER_ID = auth_resp["user_id"]
print(f"🤖 Bot user ID detected: {BOT_USER_ID}")
def fetch_recent_messages(channel_id, limit=MESSAGE_LIMIT):
try:
response = client.conversations_history(channel=channel_id, limit=limit)
messages = response["messages"]
return messages
except SlackApiError as e:
print(f"Error fetching messages: {e.response['error']}")
return []
def messages_stale(messages, minutes=5):
if not messages:
return False
latest_ts = float(messages[0]["ts"])
latest_time = datetime.fromtimestamp(latest_ts, tz=timezone.utc)
now = datetime.now(timezone.utc)
return (now - latest_time) > timedelta(minutes=minutes)
def get_last_processed_ts():
try:
with open("json/last_processed_ts.txt", "r") as f:
return float(f.read().strip())
except (FileNotFoundError, ValueError):
return 0
def save_messages(messages, path="json/messages.json", since_ts=None):
# Only keep messages not sent by the bot itself
filtered = [m for m in messages if m.get("user") != BOT_USER_ID]
# Format messages for saving (new batch)
new_batch = []
for msg in filtered:
new_batch.append({
"text": msg.get("text", ""),
"user": msg.get("user", ""),
"ts": msg.get("ts", "")
})
# Load existing messages (append mode)
try:
with open(path, "r") as f:
existing = json.load(f).get("messages", [])
except (FileNotFoundError, json.JSONDecodeError):
existing = []
# Merge and de-duplicate by ts (keep latest occurrence)
combined_map = {m.get("ts", ""): m for m in existing if m.get("ts")}
for m in new_batch:
ts = m.get("ts")
if ts:
combined_map[ts] = m
combined = list(combined_map.values())
# Sort by timestamp (newest first)
try:
combined.sort(key=lambda x: float(x.get("ts", 0)), reverse=True)
except Exception:
pass
# Create the complete messages structure
formatted = {"messages": combined}
# Save merged messages
with open(path, "w") as f:
json.dump(formatted, f, indent=2)
print(f"Saved {len(new_batch)} new messages, total now {len(combined)} (append mode)")
# Update the last processed timestamp to the latest message
if combined:
try:
latest_ts = float(combined[0].get("ts", 0))
with open("json/last_processed_ts.txt", "w") as f:
f.write(str(latest_ts))
print(f"Updated last processed timestamp to: {latest_ts}")
except Exception:
pass
else:
print("No messages found to save")
def main():
print("🔍 Fetching messages from Slack...")
messages = fetch_recent_messages(SLACK_CHANNEL_ID)
save_messages(messages)
if __name__ == "__main__":
main()
```
### kettle_monitor.py
```python
#!/usr/bin/env python3
"""
Kettle Monitor: Continuous Slack monitoring for web dashboard.
- Periodically checks Slack for new messages.
- Extracts tasks from new messages based on a silence window (no new messages for a configured duration).
- Continues Slack polling while extraction/analysis runs asynchronously.
- Designed to work with the web dashboard architecture.
"""
import os
import sys
import time
import json
import subprocess
import signal
import psutil
from datetime import datetime
import threading
import queue
import requests
import slack_fetch
import tools.extract_tasks as extract_tasks
import tools.dependency_analyzer as dependency_analyzer
import tools.project_matcher as project_matcher
# Import clear_json_files and call it at startup
from utils.json_utils import clear_json_files
clear_json_files()
os.environ["TOKENIZERS_PARALLELISM"] = "false"
class KettleMonitor:
def __init__(self):
# Polling configuration
self.poll_interval = 10.0 # seconds between Slack polls
self.silence_seconds = 60.0 # required silence (no new messages) to trigger pipeline
self.cooldown_after_run = 5.0 # seconds to wait after a run finishes
# State
self.last_slack_check = 0.0 # throttles Slack API calls
self.last_processed_ts = 0.0 # last message ts incorporated by the pipeline
self.last_new_message_ts = 0.0 # last observed newest message ts
self.pipeline_running = False # concurrency guard
# Load last processed timestamp (don't reset - run continuously)
self.load_last_processed_ts()
self.last_new_message_ts = self.last_processed_ts
self.running = True
# Lock for safe updates
self._lock = threading.Lock()
def load_last_processed_ts(self):
"""Load the timestamp of the last processed message."""
try:
if os.path.exists("json/last_processed_ts.txt"):
with open("json/last_processed_ts.txt", "r") as f:
self.last_processed_ts = float(f.read().strip())
except (ValueError, FileNotFoundError):
self.last_processed_ts = 0
def _read_messages(self):
try:
with open("json/messages.json", "r") as f:
return json.load(f).get("messages", [])
except (IOError, json.JSONDecodeError):
return []
def _update_last_processed_ts(self, ts_val: float):
self.last_processed_ts = ts_val
with open("json/last_processed_ts.txt", "w") as f:
f.write(str(self.last_processed_ts))
def _maybe_start_pipeline(self, latest_ts: float):
"""Start the pipeline in a background thread if silence is satisfied and not already running."""
now = time.time()
silence_met = (now - self.last_new_message_ts) >= self.silence_seconds
newer_than_processed = latest_ts > self.last_processed_ts
if not silence_met:
print(f"MONITOR - Silence not met ({now - self.last_new_message_ts:.1f}s < {self.silence_seconds}s); skipping run")
return
if not newer_than_processed:
print("MONITOR - No newer messages than last processed; skipping run")
return
if self.pipeline_running:
print("MONITOR - Pipeline already running; skipping run")
return
def _run_pipeline(snapshot_latest_ts: float):
with self._lock:
self.pipeline_running = True
start_t = time.time()
print("PIPELINE - Starting extract/analyze run...")
batch_path = None
try:
# Snapshot current accumulated messages into batch.json and clear messages.json for next batch
try:
with open("json/messages.json", "r") as f:
current_msgs = json.load(f).get("messages", [])
except (IOError, json.JSONDecodeError):
current_msgs = []
if current_msgs:
batch_path = "json/batch.json"
with open(batch_path, "w") as sf:
json.dump({"messages": current_msgs}, sf, indent=2)
# Clear original to let poller accumulate next batch
with open("json/messages.json", "w") as of:
json.dump({"messages": []}, of, indent=2)
# 1) Extraction
print("PIPELINE - Extracting tasks from messages (extract_tasks.main)...")
extract_tasks.main()
# 2) Dependency analysis (use categorized tasks when available)
print("PIPELINE - Running dependency analysis...")
try:
# Prefer combining coding/research tasks if present
try:
with open("json/coding_tasks.json", "r") as f:
coding_tasks = json.load(f)
except Exception:
coding_tasks = []
try:
with open("json/research_tasks.json", "r") as f:
research_tasks = json.load(f)
except Exception:
research_tasks = []
all_tasks = (coding_tasks or []) + (research_tasks or [])
if all_tasks:
dependency_analyzer.create_dependency_matrix(all_tasks)
else:
# Fallback: attempt to read phased_tasks (flat list)
try:
with open("json/phased_tasks.json", "r") as f:
phased = json.load(f)
if isinstance(phased, list):
dependency_analyzer.create_dependency_matrix(phased)
except Exception:
pass
[truncated — 3311 more characters]
```
[24 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]