mirrors/archon
1
0
Fork 0
mirror of https://github.com/coleam00/Archon.git synced 2026-01-02 04:39:29 -05:00
Code Issues Packages Projects Releases Wiki Activity

- Fix the threading service to properly handle rate limiting.

Browse Source 
- Fix the clipboard functionality to work on non local hosts and https
- Improvements in sockets on front-end and backend. Storing session in local browser storage for reconnect. Logic to prevent socket echos coausing rerender and performance issues.
- Fixes and udpates to re-ordering logic in adding a new task, reordering items on the task table.
- Allowing assignee to not be hardcoded enum.
- Fix to Document Version Control (Improvements still needed in the Milkdown editor conversion to store in the docs.
- Adding types to remove [any] typescript issues.
This commit is contained in:
sean-eskerium
2025-08-20 02:28:02 -04:00
parent cd22089b87
commit 1b5196d70f
44 changed files with 3490 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
python/src/server/api_routes/agent_chat_api.py
View File

@@ -18,9 +18,7 @@ from pydantic import BaseModel
logger = logging.getLogger(__name__)
# Import Socket.IO instance
from ..socketio_app import get_socketio_instance
sio = get_socketio_instance()
from ..socketio_app import sio
# Create router
router = APIRouter(prefix="/api/agent-chat", tags=["agent-chat"])

4
python/src/server/api_routes/knowledge_api.py
View File

@@ -35,7 +35,7 @@ from ..utils.document_processing import extract_text_from_document
# Get logger for this module
logger = get_logger(__name__)
from ..socketio_app import get_socketio_instance
from ..socketio_app import sio
from .socketio_handlers import (
complete_crawl_progress,
error_crawl_progress,
@@ -46,8 +46,6 @@ from .socketio_handlers import (
# Create router
router = APIRouter(prefix="/api", tags=["knowledge"])
# Get Socket.IO instance
sio = get_socketio_instance()
# Create a semaphore to limit concurrent crawls
# This prevents the server from becoming unresponsive during heavy crawling

4
python/src/server/api_routes/socketio_broadcasts.py
View File

@@ -8,12 +8,10 @@ No other modules should import from this file.
import asyncio
from ..config.logfire_config import get_logger
from ..socketio_app import get_socketio_instance
from ..socketio_app import sio
logger = get_logger(__name__)
# Get Socket.IO instance
sio = get_socketio_instance()
# Core broadcast functions

5
python/src/server/api_routes/socketio_handlers.py
View File

@@ -13,13 +13,10 @@ from ..config.logfire_config import get_logger
from ..services.background_task_manager import get_task_manager
from ..services.projects.project_service import ProjectService
from ..services.projects.source_linking_service import SourceLinkingService
from ..socketio_app import get_socketio_instance
from ..socketio_app import sio
logger = get_logger(__name__)
# Get Socket.IO instance
sio = get_socketio_instance()
logger.info(f"🔗 [SOCKETIO] Socket.IO instance ID: {id(sio)}")
# Rate limiting for Socket.IO broadcasts
_last_broadcast_times: dict[str, float] = {}

8
python/src/server/services/knowledge/knowledge_item_service.py
View File

@@ -128,7 +128,7 @@ class KnowledgeItemService:
code_example_counts[source_id] = 0
chunk_counts[source_id] = 0 # Default to 0 to avoid timeout
safe_logfire_info(f"Code example counts: {code_example_counts}")
safe_logfire_info("Code example counts", code_counts=code_example_counts)
# Transform sources to items with batched data
items = []
@@ -138,16 +138,18 @@ class KnowledgeItemService:
# Use batched data instead of individual queries
first_page_url = first_urls.get(source_id, f"source://{source_id}")
# Use original crawl URL instead of first page URL
original_url = source_metadata.get("original_url") or first_page_url
code_examples_count = code_example_counts.get(source_id, 0)
chunks_count = chunk_counts.get(source_id, 0)
# Determine source type
source_type = self._determine_source_type(source_metadata, first_page_url)
source_type = self._determine_source_type(source_metadata, original_url)
item = {
"id": source_id,
"title": source.get("title", source.get("summary", "Untitled")),
"url": first_page_url,
"url": original_url,
"source_id": source_id,
"code_examples": [{"count": code_examples_count}]
if code_examples_count > 0

5
python/src/server/services/projects/progress_service.py
View File

@@ -11,13 +11,10 @@ from datetime import datetime
from typing import Any
from ...config.logfire_config import get_logger
from ...socketio_app import get_socketio_instance
from ...socketio_app import sio
logger = get_logger(__name__)
# Get Socket.IO instance
sio = get_socketio_instance()
logger.info(f"🔗 [PROGRESS] Socket.IO instance ID: {id(sio)}")
class ProgressService:

4
python/src/server/services/projects/task_service.py
View File

@@ -17,9 +17,7 @@ logger = get_logger(__name__)
# Import Socket.IO instance directly to avoid circular imports
try:
from ...socketio_app import get_socketio_instance
_sio = get_socketio_instance()
from ...socketio_app import sio as _sio
_broadcast_available = True
logger.info("✅ Socket.IO broadcasting is AVAILABLE - real-time updates enabled")

7
python/src/server/services/storage/code_storage_service.py
View File

@@ -870,14 +870,17 @@ async def add_code_examples_to_supabase(
# Prepare batch data - only for successful embeddings
batch_data = []
used_indices = set() # Track which indices have been mapped to prevent duplicates
for j, (embedding, text) in enumerate(
zip(valid_embeddings, successful_texts, strict=False)
):
# Find the original index
# Find the original index (skip already used indices)
orig_idx = None
for k, orig_text in enumerate(batch_texts):
if orig_text == text:
if orig_text == text and k not in used_indices:
orig_idx = k
used_indices.add(k) # Mark this index as used
break
if orig_idx is None:

12
python/src/server/services/storage/document_storage_service.py
View File

@@ -252,20 +252,23 @@ async def add_documents_to_supabase(
search_logger.warning(
f"Skipping batch {batch_num} - no successful embeddings created"
)
completed_batches += 1
# Don't increment completed_batches when skipping - this causes progress to jump
continue
# Prepare batch data - only for successful embeddings
batch_data = []
used_indices = set() # Track which indices have been mapped to prevent duplicates
# Map successful texts back to their original indices
for j, (embedding, text) in enumerate(
zip(batch_embeddings, successful_texts, strict=False)
):
# Find the original index of this text
# Find the original index of this text (skip already used indices)
orig_idx = None
for idx, orig_text in enumerate(contextual_contents):
if orig_text == text:
if orig_text == text and idx not in used_indices:
orig_idx = idx
used_indices.add(idx) # Mark this index as used
break
if orig_idx is None:
@@ -356,6 +359,9 @@ async def add_documents_to_supabase(
search_logger.info(
f"Individual inserts: {successful_inserts}/{len(batch_data)} successful"
)
# Even if we had to fall back to individual inserts, count this batch as processed
if successful_inserts > 0:
completed_batches += 1
# Minimal delay between batches to prevent overwhelming
if i + batch_size < len(contents):

61
python/src/server/services/threading_service.py
View File

@@ -93,17 +93,18 @@ class RateLimiter:
self._clean_old_entries(now)
# Check if we can make the request
if not self._can_make_request(estimated_tokens):
while not self._can_make_request(estimated_tokens):
wait_time = self._calculate_wait_time(estimated_tokens)
if wait_time > 0:
logfire_logger.info(
f"Rate limiting: waiting {wait_time:.1f}s",
tokens=estimated_tokens,
current_usage=self._get_current_usage(),
f"Rate limiting: waiting {wait_time:.1f}s (tokens={estimated_tokens}, current_usage={self._get_current_usage()})"
)
await asyncio.sleep(wait_time)
return await self.acquire(estimated_tokens)
return False
# Clean old entries after waiting
now = time.time()
self._clean_old_entries(now)
else:
return False
# Record the request
self.request_times.append(now)
@@ -198,17 +199,13 @@ class MemoryAdaptiveDispatcher:
# Reduce workers when memory is high
workers = max(1, base // 2)
logfire_logger.warning(
"High memory usage detected, reducing workers",
memory_percent=metrics.memory_percent,
workers=workers,
f"High memory usage detected, reducing workers (memory_percent={metrics.memory_percent}, workers={workers})"
)
elif metrics.cpu_percent > self.config.cpu_threshold * 100:
# Reduce workers when CPU is high
workers = max(1, base // 2)
logfire_logger.warning(
"High CPU usage detected, reducing workers",
cpu_percent=metrics.cpu_percent,
workers=workers,
f"High CPU usage detected, reducing workers (cpu_percent={metrics.cpu_percent}, workers={workers})"
)
elif metrics.memory_percent < 50 and metrics.cpu_percent < 50:
# Increase workers when resources are available
@@ -238,12 +235,7 @@ class MemoryAdaptiveDispatcher:
semaphore = asyncio.Semaphore(optimal_workers)
logfire_logger.info(
"Starting adaptive processing",
items_count=len(items),
workers=optimal_workers,
mode=mode,
memory_percent=self.last_metrics.memory_percent,
cpu_percent=self.last_metrics.cpu_percent,
f"Starting adaptive processing (items_count={len(items)}, workers={optimal_workers}, mode={mode}, memory_percent={self.last_metrics.memory_percent}, cpu_percent={self.last_metrics.cpu_percent})"
)
# Track active workers
@@ -318,7 +310,7 @@ class MemoryAdaptiveDispatcher:
del active_workers[worker_id]
logfire_logger.error(
f"Processing failed for item {index}", error=str(e), item_index=index
f"Processing failed for item {index} (error={str(e)}, item_index={index})"
)
return None
@@ -333,11 +325,7 @@ class MemoryAdaptiveDispatcher:
success_rate = len(successful_results) / len(items) * 100
logfire_logger.info(
"Adaptive processing completed",
total_items=len(items),
successful=len(successful_results),
success_rate=f"{success_rate:.1f}%",
workers_used=optimal_workers,
f"Adaptive processing completed (total_items={len(items)}, successful={len(successful_results)}, success_rate={success_rate:.1f}%, workers_used={optimal_workers})"
)
return successful_results
@@ -355,7 +343,7 @@ class WebSocketSafeProcessor:
await websocket.accept()
self.active_connections.append(websocket)
logfire_logger.info(
"WebSocket client connected", total_connections=len(self.active_connections)
f"WebSocket client connected (total_connections={len(self.active_connections)})"
)
def disconnect(self, websocket: WebSocket):
@@ -363,7 +351,7 @@ class WebSocketSafeProcessor:
if websocket in self.active_connections:
self.active_connections.remove(websocket)
logfire_logger.info(
"WebSocket client disconnected", remaining_connections=len(self.active_connections)
f"WebSocket client disconnected (remaining_connections={len(self.active_connections)})"
)
async def broadcast_progress(self, message: dict[str, Any]):
@@ -474,7 +462,7 @@ class ThreadingService:
self._running = True
self._health_check_task = asyncio.create_task(self._health_check_loop())
logfire_logger.info("Threading service started", config=self.config.__dict__)
logfire_logger.info(f"Threading service started (config={self.config.__dict__})")
async def stop(self):
"""Stop the threading service"""
@@ -510,7 +498,7 @@ class ThreadingService:
finally:
duration = time.time() - start_time
logfire_logger.debug(
"Rate limited operation completed", duration=duration, tokens=estimated_tokens
f"Rate limited operation completed (duration={duration}, tokens={estimated_tokens})"
)
async def run_cpu_intensive(self, func: Callable, *args, **kwargs) -> Any:
@@ -562,37 +550,30 @@ class ThreadingService:
# Log system metrics
logfire_logger.info(
"System health check",
memory_percent=metrics.memory_percent,
cpu_percent=metrics.cpu_percent,
available_memory_gb=metrics.available_memory_gb,
active_threads=metrics.active_threads,
active_websockets=len(self.websocket_processor.active_connections),
f"System health check (memory_percent={metrics.memory_percent}, cpu_percent={metrics.cpu_percent}, available_memory_gb={metrics.available_memory_gb}, active_threads={metrics.active_threads}, active_websockets={len(self.websocket_processor.active_connections)})"
)
# Alert on critical thresholds
if metrics.memory_percent > 90:
logfire_logger.warning(
"Critical memory usage", memory_percent=metrics.memory_percent
f"Critical memory usage (memory_percent={metrics.memory_percent})"
)
# Force garbage collection
gc.collect()
if metrics.cpu_percent > 95:
logfire_logger.warning("Critical CPU usage", cpu_percent=metrics.cpu_percent)
logfire_logger.warning(f"Critical CPU usage (cpu_percent={metrics.cpu_percent})")
# Check for memory leaks (too many threads)
if metrics.active_threads > self.config.max_workers * 3:
logfire_logger.warning(
"High thread count detected",
active_threads=metrics.active_threads,
max_expected=self.config.max_workers * 3,
f"High thread count detected (active_threads={metrics.active_threads}, max_expected={self.config.max_workers * 3})"
)
await asyncio.sleep(self.config.health_check_interval)
except Exception as e:
logfire_logger.error("Health check failed", error=str(e))
logfire_logger.error(f"Health check failed (error={str(e)})")
await asyncio.sleep(self.config.health_check_interval)

1
python/src/server/socketio_app.py
View File

@@ -29,7 +29,6 @@ sio = socketio.AsyncServer(
# Global Socket.IO instance for use across modules
_socketio_instance: socketio.AsyncServer | None = None
def get_socketio_instance() -> socketio.AsyncServer:
"""Get the global Socket.IO server instance."""
global _socketio_instance