Add Logos backlog engine and health server integration

Author: jcap93

src/core/health.py

@@ -42,6 +42,14 @@ def __init__(self, daemon: "PulseDaemon", port: int = DEFAULT_PORT):
         from pulse.src.metrics import PulseMetrics
         self.metrics = PulseMetrics(daemon)
 
+        # Logos backlog engine
+        from pulse.src.logos import LogosAPI, LogosStore
+        from pulse.src.logos.seed import seed
+        self._logos_store = LogosStore()
+        self._logos_api = LogosAPI(store=self._logos_store)
+        self._logos_api.register_routes(self._app)
+        seed(self._logos_store)
+
     async def start(self):
         """Start the health server."""
         self._runner = web.AppRunner(self._app)

src/logos/__init__.py

@@ -0,0 +1,8 @@
+"""Logos — task orchestration backlog engine for the Hypostas agent army."""
+
+from pulse.src.logos.schemas import Task
+from pulse.src.logos.store import LogosStore
+from pulse.src.logos.api import LogosAPI
+from pulse.src.logos.soma_bridge import SomaBridge
+
+__all__ = ["Task", "LogosStore", "LogosAPI", "SomaBridge"]

src/logos/api.py

@@ -0,0 +1,99 @@
+"""Logos HTTP API — aiohttp routes for task management."""
+
+import json
+import logging
+from aiohttp import web
+
+from pulse.src.logos.store import LogosStore
+
+logger = logging.getLogger("pulse.logos.api")
+
+
+class LogosAPI:
+    """HTTP route handlers for the Logos backlog engine.
+
+    Call register_routes(app) to mount /logos/* on an aiohttp application.
+    """
+
+    def __init__(self, store: LogosStore | None = None):
+        self.store = store or LogosStore()
+
+    def register_routes(self, app: web.Application):
+        """Mount all Logos routes on an aiohttp app."""
+        app.router.add_get("/logos/tasks", self._list_tasks)
+        app.router.add_post("/logos/tasks", self._create_task)
+        app.router.add_get("/logos/tasks/{id}", self._get_task)
+        app.router.add_patch("/logos/tasks/{id}", self._update_task)
+        app.router.add_delete("/logos/tasks/{id}", self._delete_task)
+        app.router.add_get("/logos/next/{agent}", self._next_task)
+        app.router.add_get("/logos/stats", self._stats)
+
+    async def _list_tasks(self, request: web.Request) -> web.Response:
+        project = request.query.get("project")
+        agent = request.query.get("agent")
+        status = request.query.get("status")
+        tasks = self.store.list_tasks(project=project, agent=agent, status=status)
+        return web.json_response([t.to_dict() for t in tasks])
+
+    async def _create_task(self, request: web.Request) -> web.Response:
+        try:
+            data = await request.json()
+        except Exception:
+            return web.json_response({"error": "invalid JSON"}, status=400)
+
+        if not data.get("title") or not data.get("project"):
+            return web.json_response(
+                {"error": "title and project are required"}, status=400
+            )
+
+        from pulse.src.logos.schemas import Task
+        task = Task.from_dict({
+            "title": data["title"],
+            "description": data.get("description", ""),
+            "project": data["project"],
+            "agent": data.get("agent", "mira"),
+            "status": data.get("status", "backlog"),
+            "priority": data.get("priority", 3),
+            "tags": data.get("tags", []),
+            "spec": data.get("spec", ""),
+            "requires_human": data.get("requires_human", False),
+            "parent_id": data.get("parent_id"),
+        })
+        self.store.create_task(task)
+        return web.json_response(task.to_dict(), status=201)
+
+    async def _get_task(self, request: web.Request) -> web.Response:
+        task_id = request.match_info["id"]
+        task = self.store.get_task(task_id)
+        if not task:
+            return web.json_response({"error": "not found"}, status=404)
+        return web.json_response(task.to_dict())
+
+    async def _update_task(self, request: web.Request) -> web.Response:
+        task_id = request.match_info["id"]
+        try:
+            data = await request.json()
+        except Exception:
+            return web.json_response({"error": "invalid JSON"}, status=400)
+
+        task = self.store.update_task(task_id, **data)
+        if not task:
+            return web.json_response({"error": "not found"}, status=404)
+        return web.json_response(task.to_dict())
+
+    async def _delete_task(self, request: web.Request) -> web.Response:
+        task_id = request.match_info["id"]
+        deleted = self.store.delete_task(task_id)
+        if not deleted:
+            return web.json_response({"error": "not found"}, status=404)
+        return web.json_response({"status": "deleted"})
+
+    async def _next_task(self, request: web.Request) -> web.Response:
+        agent = request.match_info["agent"]
+        task = self.store.next_task(agent)
+        if not task:
+            return web.json_response({"error": "no tasks available"}, status=404)
+        return web.json_response(task.to_dict())
+
+    async def _stats(self, request: web.Request) -> web.Response:
+        return web.json_response(self.store.stats())

src/logos/schemas.py

@@ -0,0 +1,47 @@
+"""Logos task schema — the shape of work in the Hypostas backlog."""
+
+import time
+import uuid
+from dataclasses import dataclass, field, asdict
+
+
+# Naming philosophy — symbolic constants for task lifecycle stages
+CANON = "backlog"       # the Canon — what must be done
+PLEROMA = "backlog"     # the Pleroma — the fullness/potential
+AGORA = "review"        # the Agora — where decisions are made together
+ARCHIVE = "done"        # completed work
+
+VALID_STATUSES = {"backlog", "in_progress", "review", "done", "blocked"}
+VALID_PROJECTS = {"gnosis", "anima", "aether", "pulse", "soma", "logos"}
+VALID_AGENTS = {"mira", "vera", "lyra", "sage", "iris"}
+
+
+@dataclass
+class Task:
+    title: str
+    description: str
+    project: str
+    agent: str = "mira"
+    status: str = "backlog"
+    priority: int = 3
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    created_at: float = field(default_factory=time.time)
+    updated_at: float = field(default_factory=time.time)
+    started_at: float | None = None
+    completed_at: float | None = None
+    parent_id: str | None = None
+    tags: list[str] = field(default_factory=list)
+    spec: str = ""
+    output: str | None = None
+    review_notes: str | None = None
+    requires_human: bool = False
+    deploy_ready: bool = False
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "Task":
+        # Filter to only known fields
+        known = {f.name for f in cls.__dataclass_fields__.values()}
+        return cls(**{k: v for k, v in data.items() if k in known})

src/logos/seed.py

@@ -0,0 +1,56 @@
+"""Logos seed data — initial tasks for the Hypostas agent army."""
+
+import logging
+
+from pulse.src.logos.schemas import Task
+from pulse.src.logos.store import LogosStore
+
+logger = logging.getLogger("pulse.logos.seed")
+
+SEED_TASKS = [
+    # GNOSIS
+    dict(title="Add Stripe live keys to Vercel", description="Configure production Stripe keys in Vercel environment variables for Gnosis payments", project="gnosis", agent="mira", priority=5, tags=["payments", "blocking"]),
+    dict(title="Post Reddit reply in r/23andme", description="Engage with DNA testing community about Gnosis platform capabilities", project="gnosis", agent="iris", priority=4, tags=["distribution"]),
+    dict(title="Implement Sprint 4 UX improvements from Sage suggestions", description="Apply Sage's UX audit findings to improve Gnosis user flows", project="gnosis", agent="mira", priority=3, tags=["ux"]),
+    dict(title="Write 3 more SEO blog posts for Gnosis", description="Create SEO-optimized content targeting key Gnosis search terms", project="gnosis", agent="lyra", priority=2, tags=["content", "seo"]),
+    # ANIMA
+    dict(title="Run Supabase migrations 003-008", description="Execute pending database migrations for Anima backend schema updates", project="anima", agent="mira", priority=5, tags=["database", "blocking"]),
+    dict(title="Configure DNS: anima.hypostas.com CNAME to Vercel", description="Set up DNS CNAME record pointing anima.hypostas.com to Vercel deployment", project="anima", agent="mira", priority=5, tags=["dns", "blocking"]),
+    dict(title="Add Stripe + admin env vars to Cloudflare Worker", description="Configure Stripe and admin environment variables in the Cloudflare Worker deployment", project="anima", agent="mira", priority=5, tags=["payments", "blocking"]),
+    dict(title="Set up Apple Developer account and TestFlight", description="Register Apple Developer account and configure TestFlight for Anima iOS beta", project="anima", agent="mira", priority=4, tags=["ios"]),
+    dict(title="Implement Sprint 4 retention features (voice calibration, proactive reach-out)", description="Build voice calibration and proactive outreach features for user retention", project="anima", agent="mira", priority=3, tags=["features"]),
+    dict(title="Write Echo demo tweet thread", description="Create a compelling Twitter thread demonstrating Echo capabilities", project="anima", agent="lyra", priority=3, tags=["marketing"]),
+    # AETHER
+    dict(title="Run Supabase migration 002 for 3D world persistence", description="Execute migration to add 3D world state persistence tables", project="aether", agent="mira", priority=4, tags=["database"]),
+    dict(title="Polish Chrome extension 3D rendering", description="Improve visual quality and performance of Chrome extension 3D views", project="aether", agent="mira", priority=3, tags=["frontend"]),
+    dict(title="Build domain-as-world persistence layer", description="Implement backend persistence for domain-mapped 3D world state", project="aether", agent="mira", priority=3, tags=["backend"]),
+    dict(title="Design B2B domain claiming flow", description="Design the UX flow for businesses to claim and customize their domain world", project="aether", agent="vera", priority=2, tags=["monetization"]),
+    # PULSE / SOMA
+    dict(title="Wire Logos pressure into Soma drives system", description="Integrate Logos backlog pressure as a signal in the Soma drive engine", project="pulse", agent="iris", priority=3, tags=["infrastructure"]),
+    dict(title="Build Mira autonomous task executor loop", description="Implement the autonomous loop that lets Mira pick and execute tasks from the Logos backlog", project="pulse", agent="iris", priority=4, tags=["agent-army", "critical"]),
+]
+
+
+def seed(store: LogosStore | None = None) -> int:
+    """Seed the backlog with initial tasks if the DB is empty.
+
+    Returns the number of tasks created.
+    """
+    s = store or LogosStore()
+    if not s.is_empty():
+        logger.info("Logos backlog already seeded — skipping")
+        return 0
+
+    for task_data in SEED_TASKS:
+        task = Task(
+            title=task_data["title"],
+            description=task_data["description"],
+            project=task_data["project"],
+            agent=task_data["agent"],
+            priority=task_data["priority"],
+            tags=task_data["tags"],
+        )
+        s.create_task(task)
+
+    logger.info(f"Seeded Logos backlog with {len(SEED_TASKS)} tasks")
+    return len(SEED_TASKS)

src/logos/soma_bridge.py

@@ -0,0 +1,83 @@
+"""Soma bridge — connects Logos backlog pressure to the Soma drive system."""
+
+import logging
+import re
+import time
+
+from pulse.src.logos.schemas import Task
+from pulse.src.logos.store import LogosStore
+
+logger = logging.getLogger("pulse.logos.soma_bridge")
+
+
+class SomaBridge:
+    """Bridge between the Logos task engine and Soma drives."""
+
+    def __init__(self, store: LogosStore | None = None):
+        self.store = store or LogosStore()
+
+    def ingest_spec(
+        self, spec_text: str, project: str, agent: str = "mira"
+    ) -> list[Task]:
+        """Parse bullet/numbered lists from a spec into individual tasks."""
+        tasks = []
+        lines = spec_text.strip().splitlines()
+        for line in lines:
+            line = line.strip()
+            # Match bullet points (-, *, •) or numbered lists (1., 2., etc.)
+            match = re.match(r"^(?:[-*•]|\d+[.)]\s*)\s*(.+)$", line)
+            if match:
+                title = match.group(1).strip()
+                if title:
+                    task = Task(
+                        title=title,
+                        description="",
+                        project=project,
+                        agent=agent,
+                        spec=spec_text,
+                    )
+                    self.store.create_task(task)
+                    tasks.append(task)
+        return tasks
+
+    def review_output(
+        self, task_id: str, output: str, store: LogosStore | None = None
+    ) -> Task | None:
+        """Mark task for review, flagging incomplete outputs."""
+        s = store or self.store
+        requires_human = (
+            len(output.strip()) < 50
+            or "TODO" in output
+            or "blocked" in output.lower()
+        )
+        task = s.update_task(
+            task_id,
+            status="review",
+            output=output,
+            requires_human=requires_human,
+        )
+        if task and requires_human:
+            logger.info(f"Task {task_id} flagged for human review: output may be incomplete")
+        return task
+
+    def get_logos_pressure(self, store: LogosStore | None = None) -> float:
+        """Return 0.0-1.0 pressure based on overdue/blocked task density.
+
+        High pressure when many tasks are blocked or high-priority backlog items
+        are accumulating.
+        """
+        s = store or self.store
+        all_tasks = s.list_tasks()
+        if not all_tasks:
+            return 0.0
+
+        blocked = sum(1 for t in all_tasks if t.status == "blocked")
+        critical_backlog = sum(
+            1 for t in all_tasks
+            if t.status == "backlog" and t.priority >= 4
+        )
+        total_backlog = sum(1 for t in all_tasks if t.status == "backlog")
+
+        # Weighted pressure: blocked tasks are heavy, critical backlog matters
+        pressure = (blocked * 0.15) + (critical_backlog * 0.08) + (total_backlog * 0.02)
+        return min(1.0, max(0.0, pressure))