diff --git a/scripts/maintenance/branch-cleanup-delete.sh b/scripts/maintenance/branch-cleanup-delete.sh index 618019583..030237b49 100755 --- a/scripts/maintenance/branch-cleanup-delete.sh +++ b/scripts/maintenance/branch-cleanup-delete.sh @@ -13,6 +13,7 @@ # # Usage: # scripts/maintenance/branch-cleanup-delete.sh safe # delete CLOSE-SAFE branches +# scripts/maintenance/branch-cleanup-delete.sh stale # delete CLOSE-STALE branches # scripts/maintenance/branch-cleanup-delete.sh review # delete REVIEW branches (after looking) # DRY_RUN=1 scripts/maintenance/branch-cleanup-delete.sh safe # print actions only # @@ -61,6 +62,13 @@ SAFE_BRANCHES=( "v0/ultrathinking-6aaf1beb-2" ) + +# No branch currently carries the CLOSE-STALE verdict in +# docs/branch-cleanup-matrix.csv; this array intentionally starts empty and +# should be filled in from any future assessment that produces that verdict. +STALE_BRANCHES=( +) + REVIEW_BRANCHES=( "claude/create-markdown-mermaid" "claude/help-github-docs-page" @@ -89,6 +97,7 @@ archive_and_delete() { case "${1:-}" in safe) for b in "${SAFE_BRANCHES[@]}"; do archive_and_delete "$b"; done ;; + stale) for b in "${STALE_BRANCHES[@]}"; do archive_and_delete "$b"; done ;; review) for b in "${REVIEW_BRANCHES[@]}"; do archive_and_delete "$b"; done ;; - *) echo "usage: $0 {safe|review} (prefix DRY_RUN=1 to preview)"; exit 2 ;; + *) echo "usage: $0 {safe|stale|review} (prefix DRY_RUN=1 to preview)"; exit 2 ;; esac diff --git a/scripts/maintenance/session_orchestration_manager.py b/scripts/maintenance/session_orchestration_manager.py index 2f89e256b..462adaeff 100755 --- a/scripts/maintenance/session_orchestration_manager.py +++ b/scripts/maintenance/session_orchestration_manager.py @@ -10,14 +10,14 @@ import argparse import asyncio -import fnmatch import json import logging -import os -import sys from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Optional +from uuid import uuid4 + +from youtube_extension.services.shared_sql_state import SharedSQLStateStore # Set up logging logging.basicConfig( @@ -34,10 +34,23 @@ class SessionOrchestrationManager: """Programmatic API interface for playbooks, sessions, scheduling, integrations, and knowledge.""" - def __init__(self, state_path: Path = STATE_FILE_PATH): + def __init__( + self, + state_path: Path = STATE_FILE_PATH, + *, + database_url: Optional[str] = None, + ): self.state_path = state_path - self.state: Dict[str, Any] = {} + self.state: dict[str, Any] = {} self.load_state() + self._shared_state = SharedSQLStateStore( + database_url=database_url, + sqlite_path=state_path.with_suffix(".db"), + ) + legacy_sessions = self.state.get("sessions", {}) + if legacy_sessions: + self._shared_state.import_legacy_sessions(legacy_sessions) + self._refresh_sessions_from_store() def load_state(self): """Loads state from JSON, initializing with defaults if missing.""" @@ -142,10 +155,15 @@ def save_state(self): try: self.state_path.parent.mkdir(parents=True, exist_ok=True) with open(self.state_path, "w", encoding="utf-8") as f: - json.dump(self.state, f, indent=2, ensure_ascii=False) + persisted_state = dict(self.state) + persisted_state["sessions"] = {} + json.dump(persisted_state, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Failed to write state file: {e}") + def _refresh_sessions_from_store(self) -> None: + self.state["sessions"] = self._shared_state.list_sessions() + # ========================================== # Sessions API # ========================================== @@ -154,13 +172,24 @@ def create_session( self, prompt: str, playbook: str, - tags: List[str], + tags: list[str], acu_limit: int, origin: str = "user", - user: str = "jules-agent" - ) -> Dict[str, Any]: + user: str = "jules-agent", + ) -> dict[str, Any]: """Programmatically creates a new active agent session.""" - session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{len(self.state['sessions']) + 1}" + session_id = ( + f"session_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_" + f"{uuid4().hex[:8]}" + ) + initial_event = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "summary": "Session initialized", + "content": ( + f"Initialized session with playbook '{playbook}' and ACU limit " + f"{acu_limit}." + ), + } session = { "id": session_id, "prompt": prompt, @@ -171,29 +200,25 @@ def create_session( "user": user, "status": "running", "created_at": datetime.now(timezone.utc).isoformat(), - "timeline": [ - { - "timestamp": datetime.now(timezone.utc).isoformat(), - "summary": "Session initialized", - "content": f"Initialized session with playbook '{playbook}' and ACU limit {acu_limit}." - } - ] + "timeline": [initial_event], } - self.state["sessions"][session_id] = session + self._shared_state.save_session(session) + self._shared_state.append_timeline_event(session_id, initial_event) + self._refresh_sessions_from_store() self.save_state() logger.info(f"Created session {session_id} programmatically.") - return session + return self.state["sessions"][session_id] def search_sessions( self, tag: Optional[str] = None, playbook: Optional[str] = None, origin: Optional[str] = None, - user: Optional[str] = None - ) -> List[Dict[str, Any]]: + user: Optional[str] = None, + ) -> list[dict[str, Any]]: """Filters across sessions by tags, playbook, origin, or user.""" results = [] - for s in self.state["sessions"].values(): + for s in self._shared_state.list_sessions().values(): if tag and tag not in s.get("tags", []): continue if playbook and s.get("playbook") != playbook: @@ -205,9 +230,11 @@ def search_sessions( results.append(s) return results - def inspect_timeline(self, session_id: str, search_text: Optional[str] = None) -> List[Dict[str, Any]]: + def inspect_timeline( + self, session_id: str, search_text: Optional[str] = None + ) -> list[dict[str, Any]]: """Fetches the timeline event list for a session, optionally filtered by search text.""" - session = self.state["sessions"].get(session_id) + session = self._shared_state.get_session(session_id) if not session: logger.error(f"Session {session_id} not found.") return [] @@ -222,36 +249,41 @@ def inspect_timeline(self, session_id: str, search_text: Optional[str] = None) - def send_message(self, session_id: str, message: str) -> bool: """Sends a programmatic message/command to a running session, appending to timeline.""" - session = self.state["sessions"].get(session_id) + session = self._shared_state.get_session(session_id) if not session or session.get("status") != "running": logger.error(f"Session {session_id} is not active.") return False - session["timeline"].append({ + self._shared_state.append_timeline_event(session_id, { "timestamp": datetime.now(timezone.utc).isoformat(), "summary": "Received message", "content": message }) + self._refresh_sessions_from_store() self.save_state() return True def terminate_session(self, session_id: str, archive: bool = False) -> bool: """Terminates or archives an active session.""" - session = self.state["sessions"].get(session_id) + session = self._shared_state.get_session(session_id) if not session: return False session["status"] = "archived" if archive else "terminated" - session["timeline"].append({ + self._shared_state.save_session(session) + self._shared_state.append_timeline_event(session_id, { "timestamp": datetime.now(timezone.utc).isoformat(), "summary": f"Session {session['status']}", "content": f"The session was programmatically {session['status']}." }) + self._refresh_sessions_from_store() self.save_state() logger.info(f"Session {session_id} {session['status']}.") return True - async def run_parallel_sessions(self, packages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + async def run_parallel_sessions( + self, packages: list[dict[str, Any]] + ) -> list[dict[str, Any]]: """ Launches multiple sessions in parallel and waits for all of them to complete in a single async call instead of individual polling. @@ -272,12 +304,17 @@ async def run_parallel_sessions(self, packages: List[Dict[str, Any]]) -> List[Di # Auto-complete sessions post-parallel run for s in created_sessions: session_id = s["id"] - self.state["sessions"][session_id]["status"] = "completed" - self.state["sessions"][session_id]["timeline"].append({ + session = self._shared_state.get_session(session_id) + if session is None: + continue + session["status"] = "completed" + self._shared_state.save_session(session) + self._shared_state.append_timeline_event(session_id, { "timestamp": datetime.now(timezone.utc).isoformat(), "summary": "Parallel run complete", "content": "All execution items inside the playbook packages completed with success." }) + self._refresh_sessions_from_store() self.save_state() return [self.state["sessions"][s["id"]] for s in created_sessions] @@ -285,11 +322,13 @@ async def run_parallel_sessions(self, packages: List[Dict[str, Any]]) -> List[Di # Playbook Management API # ========================================== - def list_playbooks(self) -> Dict[str, Any]: + def list_playbooks(self) -> dict[str, Any]: """Lists all registered playbooks.""" return self.state["playbooks"] - def create_playbook(self, name: str, description: str, macros: List[str] = None) -> Dict[str, Any]: + def create_playbook( + self, name: str, description: str, macros: Optional[list[str]] = None + ) -> dict[str, Any]: """Creates a new automation playbook.""" playbook = { "name": name, @@ -300,7 +339,12 @@ def create_playbook(self, name: str, description: str, macros: List[str] = None) self.save_state() return playbook - def update_playbook(self, name: str, description: Optional[str] = None, macros: Optional[List[str]] = None) -> bool: + def update_playbook( + self, + name: str, + description: Optional[str] = None, + macros: Optional[list[str]] = None, + ) -> bool: """Updates an existing playbook's properties and automation macros.""" if name not in self.state["playbooks"]: return False @@ -323,7 +367,15 @@ def delete_playbook(self, name: str) -> bool: # Knowledge Management API # ========================================== - def create_knowledge_note(self, note_id: str, repo: str, folder: str, name: str, trigger: str, content: str) -> Dict[str, Any]: + def create_knowledge_note( + self, + note_id: str, + repo: str, + folder: str, + name: str, + trigger: str, + content: str, + ) -> dict[str, Any]: """Creates a new knowledge note entry.""" note = { "id": note_id, @@ -337,7 +389,9 @@ def create_knowledge_note(self, note_id: str, repo: str, folder: str, name: str, self.save_state() return note - def get_knowledge_notes(self, repo: Optional[str] = None, folder: Optional[str] = None) -> List[Dict[str, Any]]: + def get_knowledge_notes( + self, repo: Optional[str] = None, folder: Optional[str] = None + ) -> list[dict[str, Any]]: """Filters and retrieves knowledge notes.""" notes = list(self.state["knowledge_notes"].values()) if repo: @@ -354,7 +408,7 @@ def delete_knowledge_note(self, note_id: str) -> bool: return True return False - def list_suggestions(self) -> List[Dict[str, Any]]: + def list_suggestions(self) -> list[dict[str, Any]]: """Lists pending knowledge suggestions generated from sessions.""" return self.state["pending_suggestions"] @@ -371,7 +425,9 @@ def review_suggestion(self, suggestion_id: str, action: str) -> bool: # Schedule Management API # ========================================== - def create_schedule(self, schedule_id: str, cron: str, agent: str, active: bool = True) -> Dict[str, Any]: + def create_schedule( + self, schedule_id: str, cron: str, agent: str, active: bool = True + ) -> dict[str, Any]: """Creates a recurring or one-time scheduled session.""" sched = { "id": schedule_id, @@ -396,7 +452,7 @@ def toggle_schedule(self, schedule_id: str, active: bool) -> bool: # Integration Management API # ========================================== - def get_integrations(self) -> Dict[str, Any]: + def get_integrations(self) -> dict[str, Any]: """Returns the landscape of native integrations.""" return self.state["integrations"] @@ -404,7 +460,7 @@ def get_integrations(self) -> Dict[str, Any]: # Repository Documentation API # ========================================== - def search_repo_docs(self, query: str) -> List[Dict[str, str]]: + def search_repo_docs(self, query: str) -> list[dict[str, str]]: """Queries repository documentation markdown files.""" docs_dir = _PROJECT_ROOT / "docs" matches = [] diff --git a/src/youtube_extension/services/agents/adapters/agent_orchestrator.py b/src/youtube_extension/services/agents/adapters/agent_orchestrator.py index 6ffd1b345..0c53c89b1 100644 --- a/src/youtube_extension/services/agents/adapters/agent_orchestrator.py +++ b/src/youtube_extension/services/agents/adapters/agent_orchestrator.py @@ -9,13 +9,16 @@ import asyncio import logging +import os import uuid from collections import deque from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional +from ...shared_sql_state import SharedSQLStateStore from ..base_agent import AgentRequest, AgentResult, BaseAgent +from ..registry import get as get_agent_class @dataclass @@ -47,16 +50,13 @@ class OrchestrationResult: timestamp: datetime = field(default_factory=datetime.now) -from ..registry import get as get_agent_class - - class AgentOrchestrator: """ Centralized orchestration for AI agents. Handles task delegation, parallel processing, and result aggregation. """ - def __init__(self): + def __init__(self, *, database_url: Optional[str] = None): """Initialize agent orchestrator""" self.logger = logging.getLogger("agent_orchestrator") self._agents: dict[str, BaseAgent] = {} @@ -65,6 +65,21 @@ def __init__(self): # process and every dispatch appends here, so an unbounded list would # grow without limit. maxlen evicts the oldest entries automatically. self._a2a_log: deque[A2AContextMessage] = deque(maxlen=1000) + shared_database_url = ( + database_url + or os.getenv("EVENTRELAY_SHARED_STATE_DATABASE_URL") + or os.getenv("DATABASE_URL") + ) + if shared_database_url: + try: + self._shared_state = SharedSQLStateStore( + database_url=shared_database_url + ) + except Exception as exc: + self.logger.warning("Shared A2A SQL state unavailable: %s", exc) + self._shared_state = None + else: + self._shared_state = None self._task_mappings: dict[str, list[str]] = { "video_analysis": [ "video_master", @@ -78,6 +93,27 @@ def __init__(self): "strategic_analysis": ["personality_agent", "strategy_agent"], "chat_assistance": ["transcript_action"], } + self._hydrate_a2a_log() + + def _hydrate_a2a_log(self) -> None: + if self._shared_state is None: + return + self._a2a_log.clear() + for message in self._shared_state.get_a2a_messages(limit=1000): + self._a2a_log.append(A2AContextMessage(**message)) + + def _record_a2a_message(self, message: A2AContextMessage) -> None: + self._a2a_log.append(message) + if self._shared_state is not None: + self._shared_state.append_a2a_message( + { + "sender": message.sender, + "recipient": message.recipient, + "content": message.content, + "conversation_id": message.conversation_id, + "timestamp": message.timestamp, + } + ) def register_agent_type(self, name: str, agent_class: type[BaseAgent]): """ @@ -219,7 +255,7 @@ async def execute_task( content={"type": "context_share", "output": sender_result.output}, conversation_id=conv_id, ) - self._a2a_log.append(msg) + self._record_a2a_message(msg) self.logger.debug( "A2A context shared across %d agents (conv=%s)", len(orchestration_result.results), @@ -340,7 +376,7 @@ async def execute_single( ) # Record the failed dispatch so the session/audit trail is complete # (matches the success, agent-failure, and exception paths below). - self._a2a_log.append( + self._record_a2a_message( A2AContextMessage( sender="orchestrator", recipient=agent_type, @@ -360,7 +396,7 @@ async def execute_single( result = await agent.run(request) # Log execution in A2A log for session tracking - self._a2a_log.append( + self._record_a2a_message( A2AContextMessage( sender="orchestrator", recipient=agent_type, @@ -382,7 +418,7 @@ async def execute_single( return {"error": error_msg, "output": result.output} except Exception as e: self.logger.error("execute_single failed for %s: %s", agent_type, e) - self._a2a_log.append( + self._record_a2a_message( A2AContextMessage( sender="orchestrator", recipient=agent_type, @@ -416,7 +452,15 @@ def get_session_logs( List of session log entries. """ dispatch_msgs = [ - m for m in self._a2a_log + m + for m in ( + [ + A2AContextMessage(**entry) + for entry in self.get_a2a_log(limit=max(limit, 1000)) + ] + if self._shared_state is not None + else list(self._a2a_log) + ) if m.content.get("type") == "agent_dispatch" ] if agent_type: @@ -453,7 +497,7 @@ async def send_a2a_message( content=content, conversation_id=conversation_id or str(uuid.uuid4()), ) - self._a2a_log.append(msg) + self._record_a2a_message(msg) # Deliver to recipient agent if it exists agent = self._agents.get(recipient) @@ -473,6 +517,12 @@ def get_a2a_log( """Return recent A2A messages, optionally filtered by conversation.""" # Materialize to a list so `[-limit:]` slicing works (deque is not # sliceable). + if self._shared_state is not None: + return self._shared_state.get_a2a_messages( + conversation_id=conversation_id, + limit=limit, + ) + msgs = list(self._a2a_log) if conversation_id: msgs = [m for m in msgs if m.conversation_id == conversation_id] @@ -502,7 +552,7 @@ async def execute_antigravity_backend( never the input context or provider credentials. """ receipt = await backend.execute(task=task, context=context or {}) - self._a2a_log.append( + self._record_a2a_message( A2AContextMessage( sender="orchestrator", recipient="google_antigravity", diff --git a/src/youtube_extension/services/shared_sql_state.py b/src/youtube_extension/services/shared_sql_state.py new file mode 100644 index 000000000..635c91edf --- /dev/null +++ b/src/youtube_extension/services/shared_sql_state.py @@ -0,0 +1,288 @@ +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from sqlalchemy import ( + Column, + DateTime, + Integer, + MetaData, + String, + Table, + Text, + create_engine, + insert, + select, + update, +) +from sqlalchemy.engine import Engine, make_url +from sqlalchemy.exc import IntegrityError +from sqlalchemy.pool import StaticPool + +_PROJECT_ROOT = Path(__file__).resolve().parents[3] +DEFAULT_SHARED_STATE_PATH = _PROJECT_ROOT / ".runtime" / "shared_state.db" + + +def normalize_shared_storage_url(database_url: str) -> str: + """Normalize supported PostgreSQL URLs onto the Psycopg sync driver.""" + + replacements = ( + ("postgresql+asyncpg://", "postgresql+psycopg://"), + ("postgresql+psycopg2://", "postgresql+psycopg://"), + ("postgresql://", "postgresql+psycopg://"), + ("postgres://", "postgresql+psycopg://"), + ) + if database_url.startswith("postgresql+psycopg://"): + return database_url + for prefix, replacement in replacements: + if database_url.startswith(prefix): + return replacement + database_url[len(prefix) :] + return database_url + + +def resolve_shared_storage_url( + database_url: str | None = None, + *, + sqlite_path: Path | None = None, +) -> str: + """Resolve the shared state database URL from explicit, env, or local config.""" + + raw_url = ( + database_url + or os.getenv("EVENTRELAY_SHARED_STATE_DATABASE_URL") + or os.getenv("DATABASE_URL") + ) + if raw_url: + return normalize_shared_storage_url(raw_url) + + path = (sqlite_path or DEFAULT_SHARED_STATE_PATH).expanduser().resolve() + path.parent.mkdir(parents=True, exist_ok=True) + return f"sqlite:////{path.as_posix().lstrip('/')}" + + +class SharedSQLStateStore: + """Persist session timelines and A2A messages in shared SQL storage.""" + + def __init__( + self, + database_url: str | None = None, + *, + sqlite_path: Path | None = None, + ) -> None: + self.database_url = resolve_shared_storage_url( + database_url, sqlite_path=sqlite_path + ) + self.engine = self._create_engine(self.database_url) + self.metadata = MetaData() + self.sessions = Table( + "shared_sessions", + self.metadata, + Column("session_id", String(255), primary_key=True), + Column("payload", Text, nullable=False), + Column("version", Integer, nullable=False, default=1), + Column("created_at", DateTime(timezone=True), nullable=False), + Column("updated_at", DateTime(timezone=True), nullable=False), + ) + self.timeline_events = Table( + "shared_timeline_events", + self.metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("session_id", String(255), nullable=False, index=True), + Column("timestamp", String(64), nullable=False), + Column("summary", Text, nullable=False), + Column("content", Text, nullable=False), + ) + self.a2a_messages = Table( + "shared_a2a_messages", + self.metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("sender", String(255), nullable=False), + Column("recipient", String(255), nullable=False), + Column("conversation_id", String(255), nullable=False, index=True), + Column("timestamp", String(64), nullable=False), + Column("content", Text, nullable=False), + ) + self.metadata.create_all(self.engine) + + @staticmethod + def _create_engine(database_url: str) -> Engine: + url = make_url(database_url) + if url.get_backend_name() == "sqlite": + kwargs: dict[str, Any] = { + "connect_args": {"check_same_thread": False, "timeout": 30}, + "future": True, + } + if not url.database: + kwargs["poolclass"] = StaticPool + return create_engine(database_url, **kwargs) + return create_engine(database_url, future=True, pool_pre_ping=True) + + @staticmethod + def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + @staticmethod + def _serialize(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, default=str) + + @staticmethod + def _deserialize(value: str) -> dict[str, Any]: + return json.loads(value) + + def save_session(self, session: dict[str, Any]) -> dict[str, Any]: + """Create or update a session row with optimistic retries.""" + + session_id = session["id"] + payload = dict(session) + payload.pop("timeline", None) + + for _ in range(5): + now = self._utcnow() + with self.engine.begin() as conn: + current = conn.execute( + select( + self.sessions.c.payload, + self.sessions.c.version, + self.sessions.c.created_at, + ).where(self.sessions.c.session_id == session_id) + ).mappings().first() + if current is None: + try: + conn.execute( + insert(self.sessions).values( + session_id=session_id, + payload=self._serialize(payload), + version=1, + created_at=now, + updated_at=now, + ) + ) + return payload + except IntegrityError: + continue + + merged = self._deserialize(current["payload"]) + merged.update(payload) + result = conn.execute( + update(self.sessions) + .where( + self.sessions.c.session_id == session_id, + self.sessions.c.version == current["version"], + ) + .values( + payload=self._serialize(merged), + version=current["version"] + 1, + updated_at=now, + ) + ) + if result.rowcount == 1: + return merged + + raise RuntimeError(f"Failed to persist shared session state for {session_id}") + + def import_legacy_sessions(self, sessions: dict[str, dict[str, Any]]) -> None: + """Import JSON-backed sessions on first run without duplicating rows.""" + + for session in sessions.values(): + if self.get_session(session["id"]) is not None: + continue + self.save_session(session) + for event in session.get("timeline", []): + self.append_timeline_event(session["id"], event) + + def get_session(self, session_id: str) -> dict[str, Any] | None: + with self.engine.begin() as conn: + row = conn.execute( + select(self.sessions.c.payload).where(self.sessions.c.session_id == session_id) + ).scalar_one_or_none() + if row is None: + return None + session = self._deserialize(row) + session["timeline"] = self.list_timeline(session_id) + return session + + def list_sessions(self) -> dict[str, dict[str, Any]]: + with self.engine.begin() as conn: + rows = conn.execute(select(self.sessions.c.payload)).scalars().all() + sessions = {} + for payload in rows: + session = self._deserialize(payload) + session["timeline"] = self.list_timeline(session["id"]) + sessions[session["id"]] = session + return sessions + + def append_timeline_event(self, session_id: str, event: dict[str, Any]) -> None: + with self.engine.begin() as conn: + conn.execute( + insert(self.timeline_events).values( + session_id=session_id, + timestamp=event["timestamp"], + summary=event["summary"], + content=event["content"], + ) + ) + + def list_timeline(self, session_id: str) -> list[dict[str, Any]]: + with self.engine.begin() as conn: + rows = conn.execute( + select( + self.timeline_events.c.timestamp, + self.timeline_events.c.summary, + self.timeline_events.c.content, + ) + .where(self.timeline_events.c.session_id == session_id) + .order_by(self.timeline_events.c.id) + ).mappings() + return [dict(row) for row in rows] + + def append_a2a_message(self, message: dict[str, Any]) -> None: + with self.engine.begin() as conn: + conn.execute( + insert(self.a2a_messages).values( + sender=message["sender"], + recipient=message["recipient"], + conversation_id=message["conversation_id"], + timestamp=message["timestamp"], + content=self._serialize(message["content"]), + ) + ) + + def get_a2a_messages( + self, + *, + conversation_id: str | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + statement = ( + select( + self.a2a_messages.c.sender, + self.a2a_messages.c.recipient, + self.a2a_messages.c.content, + self.a2a_messages.c.conversation_id, + self.a2a_messages.c.timestamp, + ) + .order_by(self.a2a_messages.c.id.desc()) + .limit(limit) + ) + if conversation_id: + statement = statement.where( + self.a2a_messages.c.conversation_id == conversation_id + ) + with self.engine.begin() as conn: + rows = conn.execute(statement).mappings().all() + messages = [] + for row in reversed(rows): + messages.append( + { + "sender": row["sender"], + "recipient": row["recipient"], + "content": self._deserialize(row["content"]), + "conversation_id": row["conversation_id"], + "timestamp": row["timestamp"], + } + ) + return messages diff --git a/tests/unit/test_agent_orchestrator.py b/tests/unit/test_agent_orchestrator.py index 945b0b849..98ff2c677 100644 --- a/tests/unit/test_agent_orchestrator.py +++ b/tests/unit/test_agent_orchestrator.py @@ -13,6 +13,7 @@ ) from youtube_extension.services.agents.base_agent import BaseAgent from youtube_extension.services.agents.dto import AgentRequest, AgentResult +from youtube_extension.services.shared_sql_state import normalize_shared_storage_url # --------------------------------------------------------------------------- # Helpers / fakes @@ -596,6 +597,43 @@ async def test_log_entry_content_matches(self): assert entry["recipient"] == "receiver_y" assert entry["content"] == {"hello": "world"} + async def test_persists_messages_across_orchestrator_instances(self, tmp_path): + database_url = f"sqlite:///{tmp_path / 'shared-a2a.db'}" + sender = AgentOrchestrator(database_url=database_url) + receiver = AgentOrchestrator(database_url=database_url) + + await sender.send_a2a_message( + "sender_x", + "receiver_y", + {"hello": "world"}, + conversation_id="shared-conv", + ) + + log = receiver.get_a2a_log(conversation_id="shared-conv") + + assert len(log) == 1 + assert log[0]["content"] == {"hello": "world"} + + +def test_normalize_shared_storage_url_uses_psycopg_for_postgres() -> None: + postgres_url = "postgres" + "://user:pass@db.example.com/app" + psycopg2_url = "postgresql+psycopg2" + "://user:pass@db.example.com/app" + asyncpg_url = "postgresql+asyncpg" + "://user:pass@db.example.com/app" + expected = "postgresql+psycopg" + "://user:pass@db.example.com/app" + + assert ( + normalize_shared_storage_url(postgres_url) + == expected + ) + assert ( + normalize_shared_storage_url(psycopg2_url) + == expected + ) + assert ( + normalize_shared_storage_url(asyncpg_url) + == expected + ) + # =========================================================================== # execute_single diff --git a/tests/unit/test_branch_cleanup_delete_script.py b/tests/unit/test_branch_cleanup_delete_script.py new file mode 100644 index 000000000..8a1d23f5b --- /dev/null +++ b/tests/unit/test_branch_cleanup_delete_script.py @@ -0,0 +1,74 @@ +"""Regression coverage for scripts/maintenance/branch-cleanup-delete.sh. + +Locks in the fix for the ``stale`` batch, which previously fell through to +the script's ``usage: ... {safe|review}`` error path with ``exit 2`` because +``STALE_BRANCHES`` and the corresponding ``stale)`` case were missing, even +though the ``Branch Cleanup`` workflow (workflow_dispatch options and the +``[run-cleanup:stale]`` sentinel-commit regex) has always offered ``stale`` +as a selectable batch. +""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import pytest + +SCRIPT = ( + Path(__file__).resolve().parents[2] + / "scripts" + / "maintenance" + / "branch-cleanup-delete.sh" +) + + +def _run(*args: str, dry_run: bool = True) -> subprocess.CompletedProcess: + env = {"PATH": "/usr/bin:/bin"} + if dry_run: + env["DRY_RUN"] = "1" + return subprocess.run( + ["bash", str(SCRIPT), *args], + env=env, + capture_output=True, + text=True, + ) + + +@pytest.mark.skipif( + sys.platform.startswith("win"), reason="bash script not runnable on Windows" +) +class TestBranchCleanupDeleteScript: + def test_script_is_syntactically_valid(self): + result = subprocess.run( + ["bash", "-n", str(SCRIPT)], capture_output=True, text=True + ) + assert result.returncode == 0, result.stderr + + def test_stale_batch_exits_zero(self): + """Regression: `stale` must no longer hit the usage/exit-2 fallback.""" + result = _run("stale") + assert result.returncode == 0, result.stderr + + def test_safe_batch_still_exits_zero(self): + result = _run("safe") + assert result.returncode == 0, result.stderr + + def test_review_batch_still_exits_zero(self): + result = _run("review") + assert result.returncode == 0, result.stderr + + def test_unknown_batch_still_rejected_with_exit_2(self): + result = _run("bogus") + assert result.returncode == 2 + assert "usage:" in result.stdout + + def test_no_batch_still_rejected_with_exit_2(self): + result = _run() + assert result.returncode == 2 + assert "usage:" in result.stdout + + def test_usage_string_documents_stale_option(self): + result = _run("bogus") + assert "{safe|stale|review}" in result.stdout diff --git a/tests/unit/test_session_orchestration.py b/tests/unit/test_session_orchestration.py index 72a8ca0fe..6b5294d45 100644 --- a/tests/unit/test_session_orchestration.py +++ b/tests/unit/test_session_orchestration.py @@ -99,3 +99,29 @@ def test_session_orchestration_manager_knowledge_and_schedule(tmp_path): integrations = manager.get_integrations() assert "github" in integrations assert integrations["github"]["installed"] is True + + +def test_session_orchestration_manager_preserves_concurrent_timeline_updates( + tmp_path, +): + state_file = tmp_path / "shared_state.json" + manager_one = SessionOrchestrationManager(state_path=state_file) + session = manager_one.create_session( + prompt="Test concurrency", + playbook="bolt-performance-remediation", + tags=["performance"], + acu_limit=10, + ) + + manager_two = SessionOrchestrationManager(state_path=state_file) + + assert manager_one.send_message(session["id"], "first update") is True + assert manager_two.send_message(session["id"], "second update") is True + + reloaded = SessionOrchestrationManager(state_path=state_file) + timeline_contents = [ + event["content"] for event in reloaded.inspect_timeline(session["id"]) + ] + + assert "first update" in timeline_contents + assert "second update" in timeline_contents