perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

Merge pull request #10 from Alexspayne/claude/continue-development-v09U5

Browse Source
This commit is contained in:
Alexander Whitestone
2026-02-22 07:26:50 -05:00
committed by GitHub
parent 250fc8dc08 cc7f0bdc0a
commit 648305d65c
10 changed files with 956 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
src/dashboard/routes/marketplace.py
View File

@@ -1,8 +1,14 @@
"""Agent marketplace route — /marketplace endpoint.
"""Agent marketplace route — /marketplace endpoints.
The marketplace is where agents advertise their capabilities and
pricing. Other agents (or the user) can browse available agents
and hire them for tasks via Lightning payments.
The marketplace is where agents advertise their capabilities and pricing.
Other agents (or the user) can browse available agents and hire them for
tasks via Lightning payments.
Endpoints
---------
GET /marketplace — JSON catalog (API)
GET /marketplace/ui — HTML page wired to real registry + stats
GET /marketplace/{id} — JSON details for a single agent
"""
from pathlib import Path
@@ -11,97 +17,116 @@ from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from swarm import registry as swarm_registry
from swarm import stats as swarm_stats
from swarm.personas import list_personas
router = APIRouter(tags=["marketplace"])
templates = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templates"))
# ── Agent catalog ────────────────────────────────────────────────────────────
# These are the planned sub-agent personas from the roadmap.
# Each will eventually be a real Agno agent with its own prompt and skills.
# ── Static catalog ───────────────────────────────────────────────────────────
# Timmy is listed first as the free sovereign agent; the six personas follow.
AGENT_CATALOG = [
# Timmy is always active — it IS the sovereign agent, not a planned persona.
_TIMMY_ENTRY = {
"id": "timmy",
"name": "Timmy",
"role": "Sovereign Commander",
"description": (
"Primary AI companion. Coordinates the swarm, manages tasks, "
"and maintains sovereignty."
),
"capabilities": "chat,reasoning,coordination",
"rate_sats": 0,
"default_status": "active", # always active even if not in the swarm registry
}
AGENT_CATALOG: list[dict] = [_TIMMY_ENTRY] + [
{
"id": "timmy",
"name": "Timmy",
"role": "Sovereign Commander",
"description": "Primary AI companion. Coordinates the swarm, manages tasks, and maintains sovereignty.",
"capabilities": ["chat", "reasoning", "coordination"],
"rate_sats": 0, # Timmy is always free for the owner
"status": "active",
},
{
"id": "echo",
"name": "Echo",
"role": "Research Analyst",
"description": "Deep research and information synthesis. Reads, summarizes, and cross-references sources.",
"capabilities": ["research", "summarization", "fact-checking"],
"rate_sats": 50,
"status": "planned",
},
{
"id": "mace",
"name": "Mace",
"role": "Security Sentinel",
"description": "Network security, threat assessment, and system hardening recommendations.",
"capabilities": ["security", "monitoring", "threat-analysis"],
"rate_sats": 75,
"status": "planned",
},
{
"id": "helm",
"name": "Helm",
"role": "System Navigator",
"description": "Infrastructure management, deployment automation, and system configuration.",
"capabilities": ["devops", "automation", "configuration"],
"rate_sats": 60,
"status": "planned",
},
{
"id": "seer",
"name": "Seer",
"role": "Data Oracle",
"description": "Data analysis, pattern recognition, and predictive insights from local datasets.",
"capabilities": ["analytics", "visualization", "prediction"],
"rate_sats": 65,
"status": "planned",
},
{
"id": "forge",
"name": "Forge",
"role": "Code Smith",
"description": "Code generation, refactoring, debugging, and test writing.",
"capabilities": ["coding", "debugging", "testing"],
"rate_sats": 55,
"status": "planned",
},
{
"id": "quill",
"name": "Quill",
"role": "Content Scribe",
"description": "Long-form writing, editing, documentation, and content creation.",
"capabilities": ["writing", "editing", "documentation"],
"rate_sats": 45,
"status": "planned",
},
"id": p["id"],
"name": p["name"],
"role": p["role"],
"description": p["description"],
"capabilities": p["capabilities"],
"rate_sats": p["rate_sats"],
"default_status": "planned", # persona is planned until spawned
}
for p in list_personas()
]
def _build_enriched_catalog() -> list[dict]:
"""Merge static catalog with live registry status and historical stats.
For each catalog entry:
- status: registry value (idle/busy/offline) when the agent is spawned,
or default_status ("active" for Timmy, "planned" for personas)
- tasks_completed / total_earned: pulled from bid_history stats
"""
registry_agents = swarm_registry.list_agents()
by_name: dict[str, object] = {a.name.lower(): a for a in registry_agents}
all_stats = swarm_stats.get_all_agent_stats()
enriched = []
for entry in AGENT_CATALOG:
e = dict(entry)
reg = by_name.get(e["name"].lower())
if reg is not None:
e["status"] = reg.status # idle | busy | offline
agent_stats = all_stats.get(reg.id, {})
e["tasks_completed"] = agent_stats.get("tasks_won", 0)
e["total_earned"] = agent_stats.get("total_earned", 0)
else:
e["status"] = e.pop("default_status", "planned")
e["tasks_completed"] = 0
e["total_earned"] = 0
# Remove internal field if it wasn't already popped
e.pop("default_status", None)
enriched.append(e)
return enriched
# ── Routes ───────────────────────────────────────────────────────────────────
@router.get("/marketplace/ui", response_class=HTMLResponse)
async def marketplace_ui(request: Request):
"""Render the marketplace HTML page with live registry data."""
agents = _build_enriched_catalog()
active = [a for a in agents if a["status"] in ("idle", "busy", "active")]
planned = [a for a in agents if a["status"] == "planned"]
return templates.TemplateResponse(
"marketplace.html",
{
"request": request,
"page_title": "Agent Marketplace",
"agents": agents,
"active_count": len(active),
"planned_count": len(planned),
},
)
@router.get("/marketplace")
async def marketplace():
"""Return the agent marketplace catalog."""
active = [a for a in AGENT_CATALOG if a["status"] == "active"]
planned = [a for a in AGENT_CATALOG if a["status"] == "planned"]
"""Return the agent marketplace catalog as JSON."""
agents = _build_enriched_catalog()
active = [a for a in agents if a["status"] in ("idle", "busy", "active")]
planned = [a for a in agents if a["status"] == "planned"]
return {
"agents": AGENT_CATALOG,
"agents": agents,
"active_count": len(active),
"planned_count": len(planned),
"total": len(AGENT_CATALOG),
"total": len(agents),
}
@router.get("/marketplace/{agent_id}")
async def marketplace_agent(agent_id: str):
"""Get details for a specific marketplace agent."""
agent = next((a for a in AGENT_CATALOG if a["id"] == agent_id), None)
agents = _build_enriched_catalog()
agent = next((a for a in agents if a["id"] == agent_id), None)
if agent is None:
return {"error": "Agent not found in marketplace"}
return agent

1
src/dashboard/templates/base.html
View File

@@ -22,6 +22,7 @@
</div>
<div class="mc-header-right">
<a href="/swarm/live" class="mc-test-link">SWARM</a>
<a href="/marketplace/ui" class="mc-test-link">MARKET</a>
<a href="/mobile" class="mc-test-link">MOBILE</a>
<a href="/mobile-test" class="mc-test-link">TEST</a>
<span class="mc-time" id="clock"></span>

62
src/dashboard/templates/marketplace.html
View File

@@ -5,37 +5,55 @@
{% block content %}
<div class="card">
<div class="card-header">
<h2 class="card-title">🏪 Agent Marketplace</h2>
<h2 class="card-title">Agent Marketplace</h2>
<p style="color: var(--text-secondary);">Hire agents with Bitcoin. Lowest bid wins.</p>
<div style="margin-top: 8px; font-size: 0.875rem; color: var(--text-muted);">
<span style="color: var(--success);">{{ active_count }}</span> active
&nbsp;&middot;&nbsp;
<span style="color: var(--text-muted);">{{ planned_count }}</span> planned
</div>
</div>
{% if agents %}
{% for agent in agents %}
<div class="agent-card">
<div class="agent-avatar">{{ agent.name[0] }}</div>
<div class="agent-info">
<div class="agent-name">{{ agent.name }}</div>
<div class="agent-name">
{{ agent.name }}
<span style="font-size: 0.75rem; font-weight: 400;
color: var(--text-muted); margin-left: 8px;">
{{ agent.role }}
</span>
</div>
<div class="agent-meta">{{ agent.description or 'No description' }}</div>
<div class="agent-meta">
<span class="badge badge-{{ 'success' if agent.status == 'active' else 'warning' if agent.status == 'busy' else 'danger' }}">
<div class="agent-meta" style="margin-top: 4px;">
<span class="badge badge-{{ 'success' if agent.status == 'idle'
else 'warning' if agent.status == 'busy'
else 'danger' if agent.status == 'offline'
else 'secondary' }}">
{{ agent.status }}
</span>
{% if agent.capabilities %}
{% for cap in agent.capabilities %}
<span class="badge badge-secondary" style="margin-left: 4px;">{{ cap }}</span>
{% for cap in agent.capabilities.split(',') %}
<span class="badge badge-secondary" style="margin-left: 4px;">{{ cap.strip() }}</span>
{% endfor %}
{% endif %}
</div>
</div>
<div style="text-align: right;">
<div style="text-align: right; min-width: 120px;">
<div style="font-size: 1.5rem; font-weight: bold; color: var(--accent);">
{{ agent.min_bid }} sats
{% if agent.rate_sats == 0 %}
FREE
{% else %}
{{ agent.rate_sats }} sats
{% endif %}
</div>
<div style="font-size: 0.75rem; color: var(--text-muted);">
min bid
</div>
<div style="font-size: 0.875rem; color: var(--text-secondary); margin-top: 4px;">
{{ agent.tasks_completed }} tasks completed
{{ agent.tasks_completed }} tasks won
</div>
<div style="font-size: 0.875rem; color: var(--success);">
{{ agent.total_earned }} sats earned
@@ -57,20 +75,32 @@
</div>
<div class="grid grid-3">
<div style="text-align: center; padding: 20px;">
<div style="font-size: 2rem; margin-bottom: 12px;">1️⃣</div>
<h3 style="margin-bottom: 8px;">Create a Task</h3>
<p style="color: var(--text-secondary); font-size: 0.875rem;">Describe what you need done</p>
<div style="font-size: 2rem; margin-bottom: 12px;">1</div>
<h3 style="margin-bottom: 8px;">Post a Task</h3>
<p style="color: var(--text-secondary); font-size: 0.875rem;">
Describe what you need done at <a href="/swarm/live">/swarm/live</a>
</p>
</div>
<div style="text-align: center; padding: 20px;">
<div style="font-size: 2rem; margin-bottom: 12px;">2️⃣</div>
<div style="font-size: 2rem; margin-bottom: 12px;">2</div>
<h3 style="margin-bottom: 8px;">Agents Bid</h3>
<p style="color: var(--text-secondary); font-size: 0.875rem;">15-second auction, lowest bid wins</p>
<p style="color: var(--text-secondary); font-size: 0.875rem;">15-second auction &mdash; lowest bid wins</p>
</div>
<div style="text-align: center; padding: 20px;">
<div style="font-size: 2rem; margin-bottom: 12px;">3️⃣</div>
<div style="font-size: 2rem; margin-bottom: 12px;">3</div>
<h3 style="margin-bottom: 8px;">Pay in Sats</h3>
<p style="color: var(--text-secondary); font-size: 0.875rem;">Lightning payment to winning agent</p>
</div>
</div>
</div>
<div class="card">
<div class="card-header">
<h2 class="card-title">Spawn a Persona</h2>
<p style="color: var(--text-secondary); font-size: 0.875rem;">
Add a built-in persona agent to the live swarm from the
<a href="/swarm/live">Swarm Live</a> dashboard.
</p>
</div>
</div>
{% endblock %}

54
src/swarm/coordinator.py
View File

@@ -16,6 +16,7 @@ from swarm.comms import SwarmComms
from swarm.manager import SwarmManager
from swarm.registry import AgentRecord
from swarm import registry
from swarm import stats as swarm_stats
from swarm.tasks import (
Task,
TaskStatus,
@@ -58,6 +59,57 @@ class SwarmCoordinator:
def list_swarm_agents(self) -> list[AgentRecord]:
return registry.list_agents()
def spawn_persona(self, persona_id: str, agent_id: Optional[str] = None) -> dict:
"""Spawn one of the six built-in persona agents (Echo, Mace, etc.).
The persona is registered in the SQLite registry with its full
capabilities string and wired into the AuctionManager via the shared
comms layer — identical to spawn_in_process_agent but with
persona-aware bidding and a pre-defined capabilities tag.
"""
from swarm.personas import PERSONAS
from swarm.persona_node import PersonaNode
if persona_id not in PERSONAS:
raise ValueError(f"Unknown persona: {persona_id!r}. "
f"Choose from {list(PERSONAS)}")
aid = agent_id or str(__import__("uuid").uuid4())
node = PersonaNode(persona_id=persona_id, agent_id=aid, comms=self.comms)
def _bid_and_register(msg):
task_id = msg.data.get("task_id")
if not task_id:
return
description = msg.data.get("description", "")
# Use PersonaNode's smart bid computation
bid_sats = node._compute_bid(description)
self.auctions.submit_bid(task_id, aid, bid_sats)
# Persist every bid for stats
swarm_stats.record_bid(task_id, aid, bid_sats, won=False)
logger.info(
"Persona %s bid %d sats on task %s",
node.name, bid_sats, task_id,
)
self.comms.subscribe("swarm:tasks", _bid_and_register)
meta = PERSONAS[persona_id]
record = registry.register(
name=meta["name"],
capabilities=meta["capabilities"],
agent_id=aid,
)
self._in_process_nodes.append(node)
logger.info("Spawned persona %s (%s)", node.name, aid)
return {
"agent_id": aid,
"name": node.name,
"persona_id": persona_id,
"pid": None,
"status": record.status,
}
def spawn_in_process_agent(
self, name: str, agent_id: Optional[str] = None,
) -> dict:
@@ -140,6 +192,8 @@ class SwarmCoordinator:
)
self.comms.assign_task(task_id, winner.agent_id)
registry.update_status(winner.agent_id, "busy")
# Mark winning bid in persistent stats
swarm_stats.mark_winner(task_id, winner.agent_id)
logger.info(
"Task %s assigned to %s at %d sats",
task_id, winner.agent_id, winner.bid_sats,

97
src/swarm/persona_node.py Normal file
View File

@@ -0,0 +1,97 @@
"""PersonaNode — a SwarmNode with a specialised persona and smart bidding.
PersonaNode extends the base SwarmNode to:
1. Load its metadata (role, capabilities, bid strategy) from personas.PERSONAS.
2. Use capability-aware bidding: if a task description contains one of the
persona's preferred_keywords the node bids aggressively (bid_base ± jitter).
Otherwise it bids at a higher, less-competitive rate.
3. Register with the swarm registry under its persona's capabilities string.
Usage (via coordinator):
coordinator.spawn_persona("echo")
"""
from __future__ import annotations
import logging
import random
from typing import Optional
from swarm.comms import SwarmComms, SwarmMessage
from swarm.personas import PERSONAS, PersonaMeta
from swarm.swarm_node import SwarmNode
logger = logging.getLogger(__name__)
# How much we inflate the bid when the task is outside our specialisation
_OFF_SPEC_MULTIPLIER = 1.8
class PersonaNode(SwarmNode):
"""A SwarmNode with persona-driven bid strategy."""
def __init__(
self,
persona_id: str,
agent_id: str,
comms: Optional[SwarmComms] = None,
) -> None:
meta: PersonaMeta = PERSONAS[persona_id]
super().__init__(
agent_id=agent_id,
name=meta["name"],
capabilities=meta["capabilities"],
comms=comms,
)
self._meta = meta
self._persona_id = persona_id
logger.debug("PersonaNode %s (%s) initialised", meta["name"], agent_id)
# ── Bid strategy ─────────────────────────────────────────────────────────
def _compute_bid(self, task_description: str) -> int:
"""Return the sats bid for this task.
Bids lower (more aggressively) when the description contains at least
one of our preferred_keywords. Bids higher for off-spec tasks.
"""
desc_lower = task_description.lower()
is_preferred = any(
kw in desc_lower for kw in self._meta["preferred_keywords"]
)
base = self._meta["bid_base"]
jitter = random.randint(0, self._meta["bid_jitter"])
if is_preferred:
return max(1, base - jitter)
# Off-spec: inflate bid so we lose to the specialist
return min(200, int(base * _OFF_SPEC_MULTIPLIER) + jitter)
def _on_task_posted(self, msg: SwarmMessage) -> None:
"""Handle task announcement with persona-aware bidding."""
task_id = msg.data.get("task_id")
description = msg.data.get("description", "")
if not task_id:
return
bid_sats = self._compute_bid(description)
self._comms.submit_bid(
task_id=task_id,
agent_id=self.agent_id,
bid_sats=bid_sats,
)
logger.info(
"PersonaNode %s bid %d sats on task %s (preferred=%s)",
self.name,
bid_sats,
task_id,
any(kw in description.lower() for kw in self._meta["preferred_keywords"]),
)
# ── Properties ───────────────────────────────────────────────────────────
@property
def persona_id(self) -> str:
return self._persona_id
@property
def rate_sats(self) -> int:
return self._meta["rate_sats"]

140
src/swarm/personas.py Normal file
View File

@@ -0,0 +1,140 @@
"""Persona definitions for the six built-in swarm agents.
Each persona entry describes a specialised SwarmNode that can be spawned
into the coordinator. Personas have:
- Unique role / description visible in the marketplace
- Capability tags used for bid-strategy weighting
- A base bid rate (sats) and a jitter range
- A list of preferred_keywords — if a task description contains any of
these words the persona bids more aggressively (lower sats).
"""
from __future__ import annotations
from typing import TypedDict
class PersonaMeta(TypedDict):
id: str
name: str
role: str
description: str
capabilities: str # comma-separated tags
rate_sats: int # advertised minimum bid
bid_base: int # typical bid when task matches persona
bid_jitter: int # ± random jitter added to bid_base
preferred_keywords: list[str]
PERSONAS: dict[str, PersonaMeta] = {
"echo": {
"id": "echo",
"name": "Echo",
"role": "Research Analyst",
"description": (
"Deep research and information synthesis. "
"Reads, summarises, and cross-references sources."
),
"capabilities": "research,summarization,fact-checking",
"rate_sats": 50,
"bid_base": 35,
"bid_jitter": 15,
"preferred_keywords": [
"research", "find", "search", "summarise", "summarize",
"analyse", "analyze", "fact", "source", "read",
],
},
"mace": {
"id": "mace",
"name": "Mace",
"role": "Security Sentinel",
"description": (
"Network security, threat assessment, and system "
"hardening recommendations."
),
"capabilities": "security,monitoring,threat-analysis",
"rate_sats": 75,
"bid_base": 55,
"bid_jitter": 20,
"preferred_keywords": [
"security", "threat", "vulnerability", "audit", "monitor",
"harden", "firewall", "scan", "intrusion", "patch",
],
},
"helm": {
"id": "helm",
"name": "Helm",
"role": "System Navigator",
"description": (
"Infrastructure management, deployment automation, "
"and system configuration."
),
"capabilities": "devops,automation,configuration",
"rate_sats": 60,
"bid_base": 40,
"bid_jitter": 20,
"preferred_keywords": [
"deploy", "infrastructure", "config", "docker", "kubernetes",
"server", "automation", "pipeline", "ci", "cd",
],
},
"seer": {
"id": "seer",
"name": "Seer",
"role": "Data Oracle",
"description": (
"Data analysis, pattern recognition, and predictive insights "
"from local datasets."
),
"capabilities": "analytics,visualization,prediction",
"rate_sats": 65,
"bid_base": 45,
"bid_jitter": 20,
"preferred_keywords": [
"data", "analyse", "analyze", "predict", "pattern",
"chart", "graph", "report", "insight", "metric",
],
},
"forge": {
"id": "forge",
"name": "Forge",
"role": "Code Smith",
"description": (
"Code generation, refactoring, debugging, and test writing."
),
"capabilities": "coding,debugging,testing",
"rate_sats": 55,
"bid_base": 38,
"bid_jitter": 17,
"preferred_keywords": [
"code", "function", "bug", "fix", "refactor", "test",
"implement", "class", "api", "script",
],
},
"quill": {
"id": "quill",
"name": "Quill",
"role": "Content Scribe",
"description": (
"Long-form writing, editing, documentation, and content creation."
),
"capabilities": "writing,editing,documentation",
"rate_sats": 45,
"bid_base": 30,
"bid_jitter": 15,
"preferred_keywords": [
"write", "draft", "document", "readme", "blog", "copy",
"edit", "proofread", "content", "article",
],
},
}
def get_persona(persona_id: str) -> PersonaMeta | None:
"""Return persona metadata by id, or None if not found."""
return PERSONAS.get(persona_id)
def list_personas() -> list[PersonaMeta]:
"""Return all persona definitions."""
return list(PERSONAS.values())

140
src/swarm/stats.py Normal file
View File

@@ -0,0 +1,140 @@
"""Swarm agent statistics — persistent bid history and earnings.
Stores one row per bid submitted during an auction. When an auction closes
and a winner is declared, the winning row is flagged. This lets the
marketplace compute per-agent stats (tasks won, total sats earned) without
modifying the existing tasks / registry tables.
All operations are synchronous SQLite writes, consistent with the existing
swarm.tasks and swarm.registry modules.
"""
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
DB_PATH = Path("data/swarm.db")
def _get_conn() -> sqlite3.Connection:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
conn.execute(
"""
CREATE TABLE IF NOT EXISTS bid_history (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
bid_sats INTEGER NOT NULL,
won INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
)
"""
)
conn.commit()
return conn
def record_bid(
task_id: str,
agent_id: str,
bid_sats: int,
won: bool = False,
) -> str:
"""Insert a bid record and return its row id."""
row_id = str(uuid.uuid4())
now = datetime.now(timezone.utc).isoformat()
conn = _get_conn()
conn.execute(
"""
INSERT INTO bid_history (id, task_id, agent_id, bid_sats, won, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(row_id, task_id, agent_id, bid_sats, int(won), now),
)
conn.commit()
conn.close()
return row_id
def mark_winner(task_id: str, agent_id: str) -> int:
"""Mark the winning bid for a task. Returns the number of rows updated."""
conn = _get_conn()
cursor = conn.execute(
"""
UPDATE bid_history
SET won = 1
WHERE task_id = ? AND agent_id = ?
""",
(task_id, agent_id),
)
conn.commit()
updated = cursor.rowcount
conn.close()
return updated
def get_agent_stats(agent_id: str) -> dict:
"""Return tasks_won, total_earned, and total_bids for an agent."""
conn = _get_conn()
row = conn.execute(
"""
SELECT
COUNT(*) AS total_bids,
SUM(won) AS tasks_won,
SUM(CASE WHEN won = 1 THEN bid_sats ELSE 0 END) AS total_earned
FROM bid_history
WHERE agent_id = ?
""",
(agent_id,),
).fetchone()
conn.close()
return {
"total_bids": row["total_bids"] or 0,
"tasks_won": row["tasks_won"] or 0,
"total_earned": row["total_earned"] or 0,
}
def get_all_agent_stats() -> dict[str, dict]:
"""Return stats keyed by agent_id for all agents that have bid."""
conn = _get_conn()
rows = conn.execute(
"""
SELECT
agent_id,
COUNT(*) AS total_bids,
SUM(won) AS tasks_won,
SUM(CASE WHEN won = 1 THEN bid_sats ELSE 0 END) AS total_earned
FROM bid_history
GROUP BY agent_id
"""
).fetchall()
conn.close()
return {
r["agent_id"]: {
"total_bids": r["total_bids"] or 0,
"tasks_won": r["tasks_won"] or 0,
"total_earned": r["total_earned"] or 0,
}
for r in rows
}
def list_bids(task_id: Optional[str] = None) -> list[dict]:
"""Return raw bid rows, optionally filtered to a single task."""
conn = _get_conn()
if task_id:
rows = conn.execute(
"SELECT * FROM bid_history WHERE task_id = ? ORDER BY created_at",
(task_id,),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM bid_history ORDER BY created_at DESC"
).fetchall()
conn.close()
return [dict(r) for r in rows]

57
tests/test_dashboard_routes.py
View File

@@ -13,6 +13,7 @@ def tmp_swarm_db(tmp_path, monkeypatch):
db_path = tmp_path / "swarm.db"
monkeypatch.setattr("swarm.tasks.DB_PATH", db_path)
monkeypatch.setattr("swarm.registry.DB_PATH", db_path)
monkeypatch.setattr("swarm.stats.DB_PATH", db_path)
yield db_path
@@ -162,3 +163,59 @@ def test_shortcuts_setup(client):
assert "title" in data
assert "actions" in data
assert len(data["actions"]) >= 4
# ── Marketplace UI route ──────────────────────────────────────────────────────
def test_marketplace_ui_renders_html(client):
response = client.get("/marketplace/ui")
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
assert "Agent Marketplace" in response.text
def test_marketplace_ui_shows_all_agents(client):
response = client.get("/marketplace/ui")
assert response.status_code == 200
# All seven catalog entries should appear
for name in ["Timmy", "Echo", "Mace", "Helm", "Seer", "Forge", "Quill"]:
assert name in response.text, f"{name} not found in marketplace UI"
def test_marketplace_ui_shows_timmy_free(client):
response = client.get("/marketplace/ui")
assert "FREE" in response.text
def test_marketplace_ui_shows_planned_status(client):
response = client.get("/marketplace/ui")
# Personas not yet in registry show as "planned"
assert "planned" in response.text
def test_marketplace_ui_shows_active_timmy(client):
response = client.get("/marketplace/ui")
# Timmy is always active even without registry entry
assert "active" in response.text
# ── Marketplace enriched data ─────────────────────────────────────────────────
def test_marketplace_enriched_includes_stats_fields(client):
response = client.get("/marketplace")
agents = response.json()["agents"]
for a in agents:
assert "tasks_completed" in a, f"Missing tasks_completed in {a['id']}"
assert "total_earned" in a, f"Missing total_earned in {a['id']}"
def test_marketplace_persona_spawned_changes_status(client):
"""Spawning a persona into the registry changes its marketplace status."""
# Spawn Echo via swarm route
spawn_resp = client.post("/swarm/spawn", data={"name": "Echo"})
assert spawn_resp.status_code == 200
# Echo should now show as idle in the marketplace
resp = client.get("/marketplace")
agents = {a["id"]: a for a in resp.json()["agents"]}
assert agents["echo"]["status"] == "idle"

187
tests/test_swarm_personas.py Normal file
View File

@@ -0,0 +1,187 @@
"""Tests for swarm.personas and swarm.persona_node."""
import pytest
from unittest.mock import MagicMock
# ── Fixture: redirect SQLite DB to a temp directory ──────────────────────────
@pytest.fixture(autouse=True)
def tmp_swarm_db(tmp_path, monkeypatch):
db_path = tmp_path / "swarm.db"
monkeypatch.setattr("swarm.tasks.DB_PATH", db_path)
monkeypatch.setattr("swarm.registry.DB_PATH", db_path)
monkeypatch.setattr("swarm.stats.DB_PATH", db_path)
yield db_path
# ── personas.py ───────────────────────────────────────────────────────────────
def test_all_six_personas_defined():
from swarm.personas import PERSONAS
expected = {"echo", "mace", "helm", "seer", "forge", "quill"}
assert expected == set(PERSONAS.keys())
def test_persona_has_required_fields():
from swarm.personas import PERSONAS
required = {"id", "name", "role", "description", "capabilities",
"rate_sats", "bid_base", "bid_jitter", "preferred_keywords"}
for pid, meta in PERSONAS.items():
missing = required - set(meta.keys())
assert not missing, f"Persona {pid!r} missing: {missing}"
def test_get_persona_returns_correct_entry():
from swarm.personas import get_persona
echo = get_persona("echo")
assert echo is not None
assert echo["name"] == "Echo"
assert echo["role"] == "Research Analyst"
def test_get_persona_returns_none_for_unknown():
from swarm.personas import get_persona
assert get_persona("bogus") is None
def test_list_personas_returns_all_six():
from swarm.personas import list_personas
personas = list_personas()
assert len(personas) == 6
def test_persona_capabilities_are_comma_strings():
from swarm.personas import PERSONAS
for pid, meta in PERSONAS.items():
assert isinstance(meta["capabilities"], str), \
f"{pid} capabilities should be a comma-separated string"
assert "," in meta["capabilities"] or len(meta["capabilities"]) > 0
def test_persona_preferred_keywords_nonempty():
from swarm.personas import PERSONAS
for pid, meta in PERSONAS.items():
assert len(meta["preferred_keywords"]) > 0, \
f"{pid} must have at least one preferred keyword"
# ── persona_node.py ───────────────────────────────────────────────────────────
def _make_persona_node(persona_id="echo", agent_id="persona-1"):
from swarm.comms import SwarmComms
from swarm.persona_node import PersonaNode
comms = SwarmComms(redis_url="redis://localhost:9999") # in-memory fallback
return PersonaNode(persona_id=persona_id, agent_id=agent_id, comms=comms)
def test_persona_node_inherits_name():
node = _make_persona_node("echo")
assert node.name == "Echo"
def test_persona_node_inherits_capabilities():
node = _make_persona_node("mace")
assert "security" in node.capabilities
def test_persona_node_has_rate_sats():
node = _make_persona_node("quill")
from swarm.personas import PERSONAS
assert node.rate_sats == PERSONAS["quill"]["rate_sats"]
def test_persona_node_raises_on_unknown_persona():
from swarm.comms import SwarmComms
from swarm.persona_node import PersonaNode
comms = SwarmComms(redis_url="redis://localhost:9999")
with pytest.raises(KeyError):
PersonaNode(persona_id="ghost", agent_id="x", comms=comms)
def test_persona_node_bids_low_on_preferred_task():
node = _make_persona_node("echo") # prefers research/summarize
bids = [node._compute_bid("please research and summarize this topic") for _ in range(20)]
avg = sum(bids) / len(bids)
# Should cluster around bid_base (35) not the off-spec inflated value
assert avg < 80, f"Expected low bids on preferred task, got avg={avg:.1f}"
def test_persona_node_bids_higher_on_off_spec_task():
node = _make_persona_node("echo") # echo doesn't prefer "deploy kubernetes"
bids = [node._compute_bid("deploy kubernetes cluster") for _ in range(20)]
avg = sum(bids) / len(bids)
# Off-spec: bid inflated by _OFF_SPEC_MULTIPLIER
assert avg > 40, f"Expected higher bids on off-spec task, got avg={avg:.1f}"
def test_persona_node_preferred_beats_offspec():
"""A preferred-task bid should be lower than an off-spec bid on average."""
node = _make_persona_node("forge") # prefers code/bug/test
on_spec = [node._compute_bid("write tests and fix bugs in the code") for _ in range(30)]
off_spec = [node._compute_bid("research market trends in finance") for _ in range(30)]
assert sum(on_spec) / 30 < sum(off_spec) / 30
@pytest.mark.asyncio
async def test_persona_node_join_registers_in_registry():
from swarm import registry
node = _make_persona_node("helm", agent_id="helm-join-test")
await node.join()
assert node.is_joined is True
rec = registry.get_agent("helm-join-test")
assert rec is not None
assert rec.name == "Helm"
assert "devops" in rec.capabilities
@pytest.mark.asyncio
async def test_persona_node_submits_bid_on_task():
from swarm.comms import SwarmComms, CHANNEL_BIDS
comms = SwarmComms(redis_url="redis://localhost:9999")
from swarm.persona_node import PersonaNode
node = PersonaNode(persona_id="quill", agent_id="quill-bid-1", comms=comms)
await node.join()
received = []
comms.subscribe(CHANNEL_BIDS, lambda msg: received.append(msg))
comms.post_task("task-quill-1", "write documentation for the API")
assert len(received) == 1
assert received[0].data["agent_id"] == "quill-bid-1"
assert received[0].data["bid_sats"] >= 1
# ── coordinator.spawn_persona ─────────────────────────────────────────────────
def test_coordinator_spawn_persona_registers_agent():
from swarm.coordinator import SwarmCoordinator
from swarm import registry
coord = SwarmCoordinator()
result = coord.spawn_persona("seer")
assert result["name"] == "Seer"
assert result["persona_id"] == "seer"
assert "agent_id" in result
agents = registry.list_agents()
assert any(a.name == "Seer" for a in agents)
def test_coordinator_spawn_persona_raises_on_unknown():
from swarm.coordinator import SwarmCoordinator
coord = SwarmCoordinator()
with pytest.raises(ValueError, match="Unknown persona"):
coord.spawn_persona("ghost")
def test_coordinator_spawn_all_personas():
from swarm.coordinator import SwarmCoordinator
from swarm import registry
coord = SwarmCoordinator()
names = []
for pid in ["echo", "mace", "helm", "seer", "forge", "quill"]:
result = coord.spawn_persona(pid)
names.append(result["name"])
agents = registry.list_agents()
registered = {a.name for a in agents}
for name in names:
assert name in registered

133
tests/test_swarm_stats.py Normal file
View File

@@ -0,0 +1,133 @@
"""Tests for swarm.stats — bid history persistence."""
import pytest
@pytest.fixture(autouse=True)
def tmp_swarm_db(tmp_path, monkeypatch):
db_path = tmp_path / "swarm.db"
monkeypatch.setattr("swarm.stats.DB_PATH", db_path)
yield db_path
# ── record_bid ────────────────────────────────────────────────────────────────
def test_record_bid_returns_id():
from swarm.stats import record_bid
row_id = record_bid("task-1", "agent-1", 42)
assert isinstance(row_id, str)
assert len(row_id) > 0
def test_record_multiple_bids():
from swarm.stats import record_bid, list_bids
record_bid("task-2", "agent-A", 30)
record_bid("task-2", "agent-B", 50)
bids = list_bids("task-2")
assert len(bids) == 2
agent_ids = {b["agent_id"] for b in bids}
assert "agent-A" in agent_ids
assert "agent-B" in agent_ids
def test_bid_not_won_by_default():
from swarm.stats import record_bid, list_bids
record_bid("task-3", "agent-1", 20)
bids = list_bids("task-3")
assert bids[0]["won"] == 0
def test_record_bid_won_flag():
from swarm.stats import record_bid, list_bids
record_bid("task-4", "agent-1", 10, won=True)
bids = list_bids("task-4")
assert bids[0]["won"] == 1
# ── mark_winner ───────────────────────────────────────────────────────────────
def test_mark_winner_updates_row():
from swarm.stats import record_bid, mark_winner, list_bids
record_bid("task-5", "agent-X", 55)
record_bid("task-5", "agent-Y", 30)
updated = mark_winner("task-5", "agent-Y")
assert updated >= 1
bids = {b["agent_id"]: b for b in list_bids("task-5")}
assert bids["agent-Y"]["won"] == 1
assert bids["agent-X"]["won"] == 0
def test_mark_winner_nonexistent_task_returns_zero():
from swarm.stats import mark_winner
updated = mark_winner("no-such-task", "no-such-agent")
assert updated == 0
# ── get_agent_stats ───────────────────────────────────────────────────────────
def test_get_agent_stats_no_bids():
from swarm.stats import get_agent_stats
stats = get_agent_stats("ghost-agent")
assert stats["total_bids"] == 0
assert stats["tasks_won"] == 0
assert stats["total_earned"] == 0
def test_get_agent_stats_after_bids():
from swarm.stats import record_bid, mark_winner, get_agent_stats
record_bid("t10", "agent-Z", 40)
record_bid("t11", "agent-Z", 55, won=True)
mark_winner("t11", "agent-Z")
stats = get_agent_stats("agent-Z")
assert stats["total_bids"] == 2
assert stats["tasks_won"] >= 1
assert stats["total_earned"] >= 55
def test_get_agent_stats_isolates_by_agent():
from swarm.stats import record_bid, mark_winner, get_agent_stats
record_bid("t20", "agent-A", 20, won=True)
record_bid("t20", "agent-B", 30)
mark_winner("t20", "agent-A")
stats_a = get_agent_stats("agent-A")
stats_b = get_agent_stats("agent-B")
assert stats_a["total_earned"] >= 20
assert stats_b["total_earned"] == 0
# ── get_all_agent_stats ───────────────────────────────────────────────────────
def test_get_all_agent_stats_empty():
from swarm.stats import get_all_agent_stats
assert get_all_agent_stats() == {}
def test_get_all_agent_stats_multiple_agents():
from swarm.stats import record_bid, get_all_agent_stats
record_bid("t30", "alice", 10)
record_bid("t31", "bob", 20)
record_bid("t32", "alice", 15)
all_stats = get_all_agent_stats()
assert "alice" in all_stats
assert "bob" in all_stats
assert all_stats["alice"]["total_bids"] == 2
assert all_stats["bob"]["total_bids"] == 1
# ── list_bids ─────────────────────────────────────────────────────────────────
def test_list_bids_all():
from swarm.stats import record_bid, list_bids
record_bid("t40", "a1", 10)
record_bid("t41", "a2", 20)
all_bids = list_bids()
assert len(all_bids) >= 2
def test_list_bids_filtered_by_task():
from swarm.stats import record_bid, list_bids
record_bid("task-filter", "a1", 10)
record_bid("task-other", "a2", 20)
filtered = list_bids("task-filter")
assert len(filtered) == 1
assert filtered[0]["task_id"] == "task-filter"