commit d0fa193f7928699167c1dc7953a7f8ead82c3675 Author: Songhaoz666 Date: Mon Jun 8 17:32:34 2026 +0800 Initial commit diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS new file mode 100644 index 0000000..d81aed5 --- /dev/null +++ b/.github/CODEOWNERS @@ -0,0 +1,12 @@ +# CODEOWNERS — 代码所有权与 PR 审查责任绑定 +# 这些条目要求对应路径的改动必须经所有者 review。 + +# 默认：整个 Swarm 仓库 +* @Songhaoz666 + +# 文档 / 标准 +/docs/ @Songhaoz666 +/README.md @Songhaoz666 + +# Manager ↔ Swarm 契约相关（如本仓后续新增 docs/integration） +/docs/integration/ @Songhaoz666 diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..6c7dd19 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,43 @@ +## 变更说明 + + + +## 影响范围 + +- [ ] Client +- [ ] Manager +- [ ] Agent / Swarm +- [ ] CodeGW +- [ ] 计费 +- [ ] 密钥 / secret_ref +- [ ] 审计 +- [ ] 发布链路 +- [ ] 文档 / 标准 + +## 是否读取标准 + +- [ ] 已读取当前仓 CLAUDE.md +- [ ] 已读取当前仓 PROJECT_STANDARD.md +- [ ] 已读取 heicodeDocs 相关标准 + +## 是否涉及接口契约 + +- [ ] 不涉及 +- [ ] 涉及，已更新 docs/integration 或相关文档 + +## 验收方式 + + + +## TODO / Mock + +- [ ] 没有 TODO / mock +- [ ] 有，说明原因和后续计划 + +## Release 仓库确认 + +- [ ] 本 PR 不涉及 release 仓库业务代码 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6ba2abc --- /dev/null +++ b/.gitignore @@ -0,0 +1,48 @@ +# Secrets / credentials — never commit +.env +.env.* +secrets/ +*.pem +*.key +*.p12 +*.pfx +id_rsa +id_ed25519 +*_rsa +*_ed25519 + +# Python +__pycache__/ +*.py[cod] +*.pyd +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +.coverage.* + +# Virtual environments +.venv/ +venv/ +env/ +ENV/ + +# Node / desktop-client +node_modules/ +desktop-client/dist/ +desktop-client/out/ + +# Build / dist +build/ +dist/ +*.egg-info/ + +# Runtime / temp outputs +tmp-workspace/ +*.log + +# OS / editor +.DS_Store +Thumbs.db +.idea/ +.vscode/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..532ead4 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,36 @@ +# CLAUDE.md — HeiCode Swarm（执行面 / 运行时） + +本仓库（`agent_swarm_v5`，对应 **HeiCode-Swarm**）是蜂群执行面 / 回调 / Swarm Runtime。整体介绍见 [README.md](README.md)，工程标准见 [PROJECT_STANDARD.md](PROJECT_STANDARD.md)，交付说明见 [docs/DELIVERY.md](docs/DELIVERY.md)。 + +## 改动前必读 +1. 先读本 `CLAUDE.md` 与本仓 `PROJECT_STANDARD.md`。 +2. **heicodeDocs 是唯一标准源**；涉及产品/工程/Agent/Swarm/计费/安全/交付标准时，先读取 heicodeDocs 对应标准文件。 +3. 跨端或动 Manager ↔ Swarm 接口时，必须先读 `heicode-mananger/docs/heicode.md` 与 `docs/integration/` 契约。 +4. 不确定标准时，**先停止修改并要求提供标准路径**；发现文档与代码冲突时，**输出冲突点**，不得自行选择一方。 + +## 仓库边界 +- 只允许修改 Swarm（orchestrator + agent）范围；不得跨仓改 Manager、客户端、release 仓库或 heicodeDocs。 +- 一个任务只允许修改任务声明范围内的模块。 +- 禁止直接 push / force push `main`；所有改动走 PR，并按 PR 模板声明影响范围。 + +## 安全与合规 +- 禁止把密钥、Token、云凭据、`.env`、证书、私钥写入代码、日志、Markdown 或提交记录。 +- 凭据一律经部署环境 / `secret_ref` / Key Vault 注入；`.env` 已被 `.gitignore` 忽略，仅供本地开发。 +- `secret_ref`、审批链、鉴权、计费、审计相关改动必须遵循 Manager 审批链与 heicodeDocs 安全规则。 +- 禁止只写 TODO / mock / 示例代码就声称完成。 + +## 架构与关键约束（便于定位） +- **orchestrator/**：FastAPI 编排器。Manager 面接口、HMAC 签名回调、审批链**必须保持契约**。Redis 为权威存储；内存回退仅限 `REDIS_FAKE` / `ALLOW_MEMORY_STORE`（开发/CI）。 +- **agent/**：执行单元，**OpenAI 兼容**模型；保留计费/审计归属（`usage` 与 `X-Agent/X-Agnet` 头）。 +- **工作流开关默认关闭**：`ENABLE_PLANNER_FALLBACK`、`ENABLE_REVIEW_LOOP`、`ENABLE_SUBTASK_HANDOFF`。 +- 提交前必须本地通过： + ``` + python scripts/test-runtime-contract.py + python scripts/test-merge-smoke.py + python scripts/test-workflow-e2e.py + ``` + +## Agent / Teammate 协作 +- 每个 teammate / agent 必须遵循本 `CLAUDE.md` 与 heicodeDocs。 +- 不允许 teammate 跨仓自行修改未声明范围内的代码，或绕过 Manager ↔ Swarm 契约、计费、审计、审批链。 +- 多 agent 协作的最终总结需说明各 agent 负责范围、修改文件、风险与未完成项。 diff --git a/Dockerfile.agent b/Dockerfile.agent new file mode 100644 index 0000000..c8ed46b --- /dev/null +++ b/Dockerfile.agent @@ -0,0 +1,26 @@ +FROM python:3.11-slim + +# Install Git +RUN apt-get update && \ + apt-get install -y git && \ + apt-get clean && \ + rm -rf /var/lib/apt/lists/* + +# Set working directory +WORKDIR /app + +# Copy agent code +COPY agent/ /app/agent/ + +# Install Python dependencies +RUN pip install --no-cache-dir -r /app/agent/requirements.txt + +# Create workspace directory +RUN mkdir -p /workspace + +# Set environment variables +ENV PYTHONUNBUFFERED=1 +ENV WORKSPACE_DIR=/workspace + +# Run agent +CMD ["python", "-m", "agent.main"] diff --git a/Dockerfile.orchestrator b/Dockerfile.orchestrator new file mode 100644 index 0000000..4d508d0 --- /dev/null +++ b/Dockerfile.orchestrator @@ -0,0 +1,16 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Install dependencies +COPY orchestrator/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy orchestrator code +COPY orchestrator/ ./orchestrator/ + +# Expose port +EXPOSE 8000 + +# Run orchestrator +CMD ["python", "-m", "uvicorn", "orchestrator.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/PROJECT_STANDARD.md b/PROJECT_STANDARD.md new file mode 100644 index 0000000..e510a54 --- /dev/null +++ b/PROJECT_STANDARD.md @@ -0,0 +1,37 @@ +# PROJECT_STANDARD.md — HeiCode Swarm 工程标准 + +本文件约定本仓（`agent_swarm_v5` / HeiCode-Swarm）的工程标准。与 heicodeDocs 冲突时，以 **heicodeDocs** 为准；本文件仅作本仓落地补充。配套：[CLAUDE.md](CLAUDE.md)、[README.md](README.md)、[docs/DELIVERY.md](docs/DELIVERY.md)。 + +## 运行时与依赖 +- Python 3.13+；依赖见 `orchestrator/requirements.txt`、`agent/requirements.txt`。 +- 模型后端：**OpenAI 兼容** Chat Completions（`OPENAI_API_KEY` / `OPENAI_API_BASE` / `OPENAI_MODEL`，可指向自定义端点）。 +- 持久化：**Redis 为权威存储**。`fakeredis` 内存回退仅限开发/CI，且必须由 `REDIS_FAKE` / `ALLOW_MEMORY_STORE` 显式开启；生产在 Redis 不可用时快速失败。 + +## 功能开关（默认关闭，保持契约） +- `ENABLE_PLANNER_FALLBACK`：无 Manager 分工时启用 LLM 规划回退。 +- `ENABLE_REVIEW_LOOP` / `MAX_REVIEW_CYCLES`：主控评审/重做循环 + 结果汇总。 +- `ENABLE_SUBTASK_HANDOFF`：动态子任务移交。 +- 开启上述开关不得改变 Manager 面接口、回调、审批链与计费/审计字段语义。 + +## 安全 +- 不得提交密钥/Token/凭据/`.env`/证书/私钥；通过环境 / `secret_ref` / Key Vault 注入。 +- K8s 部署：Agent 模型密钥使用 Secret `openai-secret`（Pod 模板）或 `openai-api-key`（Deployment 清单）。 +- 保留模型用量归属（`usage`、`X-Agent/X-Agnet` 头）以满足计费/审计。 + +## 测试与验收（PR 前必须通过） +``` +python scripts/test-runtime-contract.py # Manager 契约校验 +python scripts/test-merge-smoke.py # 机制级冒烟（评审/协作/汇总等） +python scripts/test-workflow-e2e.py # 端到端工作流（无需密钥） +``` +- 新增/修改逻辑应补充对应断言；不得以 TODO / mock 充当完成。 +- 涉及 Manager 契约的改动，必须先通过 `test-runtime-contract.py`。 + +## 部署 +- 正式部署走**镜像**：`Dockerfile.orchestrator` / `Dockerfile.agent` + `k8s/` 清单（`orchestrator-deployment.yaml` 使用 `swarm-orchestrator:latest`）。 +- `k8s/orchestrator-source-configmap.yaml` 为陈旧快照，请勿用于当前部署（见 `docs/DELIVERY.md` §九）。 + +## 提交与分支 +- 禁止直接 push / force push `main`；所有改动经 PR 合并，至少 1 人 review。 +- PR 必须按 `.github/pull_request_template.md` 声明影响范围（Client / Manager / Agent-Swarm / CodeGW / 计费 / 密钥 / 审计 / 发布链路 / 文档）。 +- 提交信息清晰描述改动；保持与现有代码风格一致。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..0eed45c --- /dev/null +++ b/README.md @@ -0,0 +1,93 @@ +# Agent Swarm（HeiCode Swarm） + +一个多智能体「蜂群」系统：用户提出需求后，由主控逻辑自动将其分解为多个子任务，分发给擅长不同领域的专家 Agent 并行完成；专家之间可就重叠领域相互协作；产出汇总后由主控评审是否达标，未达标则退回重做，循环直至生成满意的最终回答。 + +## 系统架构 + +``` + 用户 + │ + 桌面客户端（desktop-client） + │ 提交需求 / 查看状态与结果 + ▼ + Heicode Manager（控制面，外部） + │ 下发编排方案 / 接收带签名回调（审批、计费、审计） + ▼ + Orchestrator（编排器，FastAPI） + 分解 · 派发 · 协作路由 · 评审/重做 · 汇总 + ├─ Redis（权威状态存储） + └─ WebSocket ┐ + ▼ + Agent · Agent · Agent …（执行单元 / K8s Pod，调用大模型完成任务） +``` + +## 工作流 + +``` +分解（Plan） → 派发给专家（Dispatch） → 专家执行（Execute） + → 重叠领域协作 / 移交（Collaborate & Handoff） + → 主控评审（Review）──不达标──▶ 退回相关任务重做（循环，受上限约束） + └──达标──▶ 汇总为统一回答并交付（Deliver） +``` + +- **分解**：优先采用 Manager 提供的编排方案；若未提供且开启规划回退（`ENABLE_PLANNER_FALLBACK`），由编排器用大模型把目标拆解为「实现 → 测试 → 文档」等专家子任务。 +- **派发**：按「能力匹配 + 剩余容量」将就绪任务下发给已连接的 Agent；派发时把已完成依赖的产物与同伴信息注入上下文。 +- **协作 / 移交**：测试、文档等角色可向实现角色咨询以保持语义一致；复杂子任务可移交给更合适的专家。 +- **评审 / 重做**：所有任务完成后由评审者判断是否达标（开启 `ENABLE_REVIEW_LOOP`），不达标则退回相关任务重做，受 `MAX_REVIEW_CYCLES` 约束。 +- **汇总交付**：通过后将各专家产出汇总为统一、面向用户的最终回答。 + +## 仓库结构与子文档 + +| 目录 | 角色 | 文档 | +|---|---|---| +| `orchestrator/` | 编排器：分解、派发、协作、评审、汇总、Manager 对接 | [orchestrator/README.md](orchestrator/README.md) | +| `agent/` | 执行单元：连接编排器、调用大模型完成任务、提交结果 | [agent/README.md](agent/README.md) | +| `desktop-client/` | 桌面客户端：任务提交、复杂度分析、状态展示、结果聚合 | [desktop-client/README.md](desktop-client/README.md) | +| `desktop-client/src/ai/` | 复杂度分析模块：评估规模、推荐 Agent 数量 | [desktop-client/src/ai/README.md](desktop-client/src/ai/README.md) | +| `docs/` | 交付说明（交付物、构建/部署、配置、验收、影响与合规） | [docs/DELIVERY.md](docs/DELIVERY.md) | +| `k8s/` | Kubernetes 部署清单（编排器、Agent、Redis、RBAC、监控等） | — | +| `scripts/` | 部署、契约校验、冒烟与端到端工作流测试脚本 | — | +| `test-data/` | 复杂度分析的标注数据集 | — | + +## 快速开始（本地） + +需要 Python 3.13+ 环境。本地开发可用受控的内存存储（`REDIS_FAKE=1`，免装 Redis）。 + +```bash +# 1) 编排器（在本目录 agent_swarm_v5 下运行，开启规划回退与评审循环） +set "REDIS_FAKE=1" +set "ENABLE_PLANNER_FALLBACK=1" +set "ENABLE_REVIEW_LOOP=1" +python -m uvicorn orchestrator.main:app --host 0.0.0.0 --port 8000 + +# 2) 一个 Agent（凭据放在本目录 .env：OPENAI_API_KEY 等） +set "ORCHESTRATOR_URL=ws://localhost:8000" +set "AGENT_ID=worker-1" +set "AGENT_CAPABILITIES=python,code_generation,testing,pytest,technical-writing,general" +set "WORKSPACE_DIR=..\tmp-workspace\worker-1" +python -m agent.main + +# 3) 提交一个需求 +curl -s -X POST http://localhost:8000/api/swarms -H "Content-Type: application/json" ^ + -d "{\"mode\":\"swarm\",\"requirement\":{\"objective\":\"实现 add(a,b) 并补充测试与说明\"},\"callback\":{\"url\":\"http://localhost:9999/cb\",\"subscribed_events\":[]},\"metadata\":{\"manager_deployment_id\":\"dev-1\"}}" +``` + +随后可通过 `GET /api/swarms/{deployment_id}/workflow` 与 `/logs` 观察分解、派发、评审与汇总过程。各组件的环境变量与接口详见对应子文档。 + +## 关键设计 + +- **存储**：Redis 为权威状态存储；仅 `REDIS_FAKE` / `ALLOW_MEMORY_STORE` 开启时才使用进程内回退，生产环境在 Redis 不可用时快速失败。 +- **模型**：Agent 与编排器规划/评审均使用 **OpenAI 兼容** API，可指向自定义端点；未配置密钥时规划/评审退化为静态分解与启发式判定。 +- **Manager 契约**：面向 Manager 的接口、带签名回调、审批链与计费/审计字段均予以保留。 + +## 测试 + +```bash +python scripts/test-runtime-contract.py # Manager 契约校验 +python scripts/test-merge-smoke.py # 工作流冒烟测试（评审 / 协作 / 汇总等） +``` + +## 安全与合规 + +- **禁止**将密钥、令牌、云凭据写入代码、日志或提交记录；凭据通过 `.env`（已忽略）或部署环境 / `secret_ref` 注入。 +- 涉及鉴权、审批链、计费与审计的改动需遵循 Manager 安全规则。 diff --git a/agent/README.md b/agent/README.md new file mode 100644 index 0000000..a663fa9 --- /dev/null +++ b/agent/README.md @@ -0,0 +1,104 @@ +# Agent 执行器（Agent Runtime） + +Agent 是蜂群系统中的执行单元（worker）。每个 Agent 作为独立进程运行（在 Kubernetes 中通常是一个 Pod），通过 WebSocket 连接到 Orchestrator，领取任务、调用大模型完成编码工作，并把结果写回 Git 分支。 + +## 职责概览 + +- 通过 WebSocket 连接 Orchestrator，并以自身**能力（capabilities）**注册，例如 `python`、`testing`、`technical-writing`。 +- 接收 Orchestrator 下发的任务，使用 **OpenAI 兼容**的大模型完成子任务。 +- 在隔离的「按任务工作目录」中生成/修改文件，并提交到独立的结果分支。 +- 支持与其它 Agent **协作（peer collaboration）**与**移交（handoff）**。 +- 持续上报心跳、状态与执行结果，并暴露 Prometheus 指标。 + +## 目录结构 + +``` +agent/ +├── main.py # Agent 主循环：连接、注册、心跳、消息处理、并发与重连 +├── task_executor.py # 任务执行引擎：调用 OpenAI 兼容模型，应用文件改动 +├── handoff_logic.py # 移交决策逻辑（复杂度 / 能力 / 预估时长） +├── git_operations.py # Git 克隆、建分支、提交、推送 +└── requirements.txt # Python 依赖 +``` + +## 核心模块 + +### 主循环（`main.py`） +- **连接与重连**：断线后按指数退避自动重连并重新注册。 +- **有界并发**：通过信号量限制同时执行的任务数（`MAX_CONCURRENT_TASKS`），并在注册/心跳中上报剩余容量 `available_slots`。 +- **任务受理**：对每个分配先做校验，重复任务返回 `task_accepted(status=duplicate)`，超出容量返回 `task_rejected`，由 Orchestrator 重新入队。 +- **执行保护**：单任务超时（`TASK_TIMEOUT_SECONDS`）与任务取消（`cancel_task`）。 +- **心跳**：每 15 秒上报一次存活与容量信息。 +- **优雅退出**：收到 SIGINT/SIGTERM 时停止接单并清理在执行的任务。 + +### 任务执行器（`task_executor.py`） +- 使用 **OpenAI 兼容 API**（可指向自定义 `base_url`）完成子任务。 +- 根据上下文中的**依赖产物（dependency_artifacts）**与**专家角色（specialist_role）**对齐输出：测试与文档以实现产物的行为/异常语义为准。 +- 将模型返回的文件改动安全地写入工作目录（带路径越界校验）。 +- 记录用量并附带计费/审计归属（`usage` 与 `X-Agent/X-Agnet` 模型归属头），供 Orchestrator 上报。 + +### 移交逻辑（`handoff_logic.py`） +当子任务满足以下条件时建议移交给更合适的 Agent： +- 复杂度为 `high`； +- 需要当前 Agent 不具备的专业能力； +- 预估耗时超过 60 分钟。 + +### Git 操作（`git_operations.py`） +- 自动识别仓库根目录（`repo_root`），使按任务子目录也能在父级 Git 仓库中正确执行。 +- 为每个任务创建结果分支（形如 `agent/{agent-id}/{task}-{timestamp}`），提交并推送。 + +## 配置（环境变量） + +凭据通过同目录的 `.env`（已被 `.gitignore` 忽略）加载，亦可由部署环境/`secret_ref` 注入。**请勿将密钥写入代码或提交记录。** + +```bash +# 模型（OpenAI 兼容） +OPENAI_API_KEY=... # 或 MODEL_API_KEY +OPENAI_API_BASE=https://api.openai.com/v1 # 或 MODEL_API_BASE，可指向自定义端点 +OPENAI_MODEL=gpt-4o-mini # 或 MODEL_NAME / MODEL_ID + +# 连接与身份 +ORCHESTRATOR_URL=ws://localhost:8000 # Orchestrator 的 WebSocket 地址 +AGENT_ID=worker-1 # 不填则自动生成 +AGENT_CAPABILITIES=python,testing # 逗号分隔的能力列表 +WORKSPACE_DIR=/workspace # 工作目录 + +# 可选 +GIT_REPO_URL=https://... # 需在仓库内工作时设置 +GIT_USERNAME / GIT_PASSWORD / GIT_TOKEN# 推送凭据 +GIT_BASE_BRANCH=main # 结果分支的基线 +MAX_CONCURRENT_TASKS=4 # 最大并发任务数 +TASK_TIMEOUT_SECONDS=60 # 单任务超时 +METRICS_PORT=9000 # 设置后暴露 Prometheus 指标 +ENABLE_SUBTASK_HANDOFF=false # 是否启用动态子任务移交 +``` + +## 本地运行 + +```bash +pip install -r agent/requirements.txt + +# 准备 .env（含 OPENAI_API_KEY 等），然后从仓库根目录运行： +set "ORCHESTRATOR_URL=ws://localhost:8000" +set "AGENT_ID=worker-1" +set "AGENT_CAPABILITIES=python,code_generation,testing,pytest,technical-writing,general" +set "WORKSPACE_DIR=...\tmp-workspace\worker-1" +python -m agent.main +``` + +## 任务执行流程 + +1. **注册**：连接 Orchestrator 并上报能力与可用容量。 +2. **心跳**：每 15 秒上报存活与剩余容量。 +3. **受理**：校验分配，必要时拒绝（容量不足）或忽略（重复）。 +4. **执行**：在独立的按任务工作目录中调用模型完成子任务；如启用，可向同伴 Agent 咨询或移交。 +5. **提交**：若为 Git 工作区，提交并推送到结果分支。 +6. **回报**：将结果（含用量）发送回 Orchestrator，并将状态置为空闲。 + +## 协作（Peer Collaboration） + +当上下文提供了 `peer_agents` 时，测试/文档等角色可向实现角色发起咨询；被咨询方会回复自己最近一次完成任务的摘要，帮助各专家在重叠领域保持一致。 + +## 监控 + +Agent 暴露的 Prometheus 指标包括：已执行/失败任务数、任务时长、移交次数、重连次数、被拒/重复任务数、当前活跃任务数与 Agent 状态。 diff --git a/agent/__init__.py b/agent/__init__.py new file mode 100644 index 0000000..6fa609b --- /dev/null +++ b/agent/__init__.py @@ -0,0 +1,3 @@ +"""Agent package for K8s-based swarm mode.""" + +__all__: list[str] = [] diff --git a/agent/git_operations.py b/agent/git_operations.py new file mode 100644 index 0000000..a00455e --- /dev/null +++ b/agent/git_operations.py @@ -0,0 +1,219 @@ +"""Git operations for the agent workspace. + +Merged from agent_swarm_v4: +- Adds repository-root discovery (`repo_root`) so per-task child workspaces can still + operate inside a parent Git checkout (all git commands run at the repo toplevel) +- Keeps the original branch/commit/push API so the protocol stays compatible +""" +import asyncio +import logging +import os +import shutil +import subprocess +import time +from pathlib import Path +from typing import Optional +from urllib.parse import quote, urlsplit, urlunsplit + +logger = logging.getLogger(__name__) + + +class GitOperations: + """Handles Git operations for agent workspace.""" + + def __init__(self, workspace_dir: str, agent_id: str): + self.workspace_dir = str(Path(workspace_dir)) + self.agent_id = agent_id + self.result_branch = f"agent-{agent_id}-results" + self.base_branch = os.getenv("GIT_BASE_BRANCH", "main") + + def _authenticated_repo_url(self, repo_url: str) -> str: + username = os.getenv("GIT_USERNAME") + password = os.getenv("GIT_PASSWORD") or os.getenv("GIT_TOKEN") + if not username or not password: + return repo_url + + parsed = urlsplit(repo_url) + if parsed.scheme not in {"http", "https"} or "@" in parsed.netloc: + return repo_url + + userinfo = f"{quote(username)}:{quote(password)}" + return urlunsplit(( + parsed.scheme, + f"{userinfo}@{parsed.netloc}", + parsed.path, + parsed.query, + parsed.fragment, + )) + + async def repo_root(self) -> Optional[str]: + result = await self._run_git_command( + ["git", "rev-parse", "--show-toplevel"], + cwd=self.workspace_dir, + ) + if result.returncode != 0: + return None + return result.stdout.strip() or None + + async def is_git_workspace(self) -> bool: + result = await self._run_git_command( + ["git", "rev-parse", "--is-inside-work-tree"], + cwd=self.workspace_dir, + ) + return result.returncode == 0 and result.stdout.strip() == "true" + + async def clone_workspace(self, repo_url: str) -> bool: + try: + logger.info(f"Cloning repository from {repo_url}") + + workspace = Path(self.workspace_dir) + workspace.parent.mkdir(parents=True, exist_ok=True) + + if workspace.exists(): + if await self.is_git_workspace(): + logger.info("Workspace already contains a Git checkout; refreshing existing clone") + return await self._refresh_existing_workspace(repo_url) + + if any(workspace.iterdir()): + logger.warning("Workspace directory exists and is not empty; clearing before clone") + self._clear_workspace_dir(workspace) + + clone_url = self._authenticated_repo_url(repo_url) + result = await self._run_git_command([ + "git", "clone", clone_url, self.workspace_dir + ]) + + if result.returncode != 0: + logger.error(f"Git clone failed: {result.stderr}") + return False + + logger.info(f"Successfully cloned repository to {self.workspace_dir}") + await self._configure_git_user() + return True + + except Exception as e: + logger.error(f"Error cloning workspace: {e}") + return False + + def _clear_workspace_dir(self, workspace: Path): + for child in workspace.iterdir(): + if child.is_dir() and not child.is_symlink(): + shutil.rmtree(child) + else: + child.unlink() + + async def _configure_git_user(self): + cwd = await self.repo_root() or self.workspace_dir + await self._run_git_command([ + "git", "config", "user.name", f"Agent {self.agent_id}" + ], cwd=cwd) + await self._run_git_command([ + "git", "config", "user.email", f"{self.agent_id}@agent.local" + ], cwd=cwd) + + async def _refresh_existing_workspace(self, repo_url: str) -> bool: + cwd = await self.repo_root() or self.workspace_dir + clone_url = self._authenticated_repo_url(repo_url) + await self._run_git_command(["git", "remote", "set-url", "origin", clone_url], cwd=cwd) + await self._configure_git_user() + + fetch = await self._run_git_command(["git", "fetch", "origin"], cwd=cwd) + if fetch.returncode != 0: + logger.error(f"Git fetch failed: {fetch.stderr}") + return False + + base_ref = f"origin/{self.base_branch}" + base_check = await self._run_git_command(["git", "rev-parse", "--verify", base_ref], cwd=cwd) + if base_check.returncode == 0: + reset = await self._run_git_command(["git", "reset", "--hard", base_ref], cwd=cwd) + if reset.returncode != 0: + logger.error(f"Git reset failed: {reset.stderr}") + return False + + clean = await self._run_git_command(["git", "clean", "-fd"], cwd=cwd) + if clean.returncode != 0: + logger.error(f"Git clean failed: {clean.stderr}") + return False + + return True + + async def create_result_branch(self, task_id: Optional[str] = None) -> bool: + try: + cwd = await self.repo_root() or self.workspace_dir + if task_id: + safe_task_id = task_id.replace("/", "-")[:12] + timestamp = int(time.time()) + self.result_branch = f"agent/{self.agent_id}/{safe_task_id}-{timestamp}" + + await self._run_git_command(["git", "fetch", "origin"], cwd=cwd) + + base_ref = f"origin/{self.base_branch}" + base_check = await self._run_git_command(["git", "rev-parse", "--verify", base_ref], cwd=cwd) + if base_check.returncode != 0: + base_ref = "HEAD" + logger.warning(f"Base branch origin/{self.base_branch} not found; creating result branch from HEAD") + + result = await self._run_git_command(["git", "checkout", "-B", self.result_branch, base_ref], cwd=cwd) + if result.returncode != 0: + logger.error(f"Failed to create branch: {result.stderr}") + return False + + logger.info(f"Created result branch: {self.result_branch}") + return True + except Exception as e: + logger.error(f"Error creating result branch: {e}") + return False + + async def commit_changes(self, message: str) -> Optional[str]: + try: + cwd = await self.repo_root() or self.workspace_dir + status_result = await self._run_git_command(["git", "status", "--porcelain"], cwd=cwd) + if not status_result.stdout.strip(): + logger.info("No changes to commit") + return None + + await self._run_git_command(["git", "add", "-A"], cwd=cwd) + commit_result = await self._run_git_command(["git", "commit", "-m", message], cwd=cwd) + if commit_result.returncode != 0: + logger.error(f"Git commit failed: {commit_result.stderr}") + return None + + sha_result = await self._run_git_command(["git", "rev-parse", "HEAD"], cwd=cwd) + commit_sha = sha_result.stdout.strip() + logger.info(f"Committed changes: {commit_sha[:8]} - {message}") + return commit_sha + except Exception as e: + logger.error(f"Error committing changes: {e}") + return None + + async def push_results(self) -> Optional[str]: + try: + cwd = await self.repo_root() or self.workspace_dir + result = await self._run_git_command(["git", "push", "-u", "origin", self.result_branch], cwd=cwd) + if result.returncode != 0: + logger.error(f"Git push failed: {result.stderr}") + return None + logger.info(f"Pushed results to branch: {self.result_branch}") + return self.result_branch + except Exception as e: + logger.error(f"Error pushing results: {e}") + return None + + async def _run_git_command( + self, + command: list[str], + cwd: Optional[str] = None, + ) -> subprocess.CompletedProcess: + process = await asyncio.create_subprocess_exec( + *command, + cwd=cwd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + return subprocess.CompletedProcess( + args=command, + returncode=process.returncode, + stdout=stdout.decode(), + stderr=stderr.decode(), + ) diff --git a/agent/handoff_logic.py b/agent/handoff_logic.py new file mode 100644 index 0000000..1a802b3 --- /dev/null +++ b/agent/handoff_logic.py @@ -0,0 +1,224 @@ +"""Handoff decision logic for determining when to delegate tasks.""" +import logging +from typing import Iterable, List, Optional +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass +class HandoffDecision: + """Result of handoff decision analysis.""" + should_handoff: bool + reason: str + target_capabilities: List[str] + + +CAPABILITY_ALIASES = { + "code_generation": { + "code-reading", + "code-editing", + "command-line", + "credentials-handling", + "editing", + "environment-inspection", + "environment-setup", + "file-editing", + "file-operations", + "file-system", + "filesystem", + "read-only-operations", + "reporting", + "repository-inspection", + "shell", + "testing", + "pytest", + "quality-checking", + "attention-to-detail", + "error-handling", + "technical-writing", + "validation", + "verification", + }, + "python": { + "pytest", + "testing", + "code-reading", + "code-editing", + "command-line", + "editing", + "file-editing", + "file-operations", + "file-system", + "filesystem", + "reporting", + "shell", + "quality-checking", + "error-handling", + "validation", + "verification", + }, + "general": { + "general", + "code-reading", + "file-system", + "filesystem", + "quality-checking", + "attention-to-detail", + }, +} + + +def _expanded_capabilities(capabilities: Iterable[str]) -> set[str]: + """Return capabilities plus local aliases supported by this agent.""" + expanded = {"general", *capabilities} + for capability in capabilities: + expanded.update(CAPABILITY_ALIASES.get(capability, set())) + return expanded + + +def should_handoff( + subtask: dict, + agent_id: str, + agent_capabilities: Optional[List[str]] = None, +) -> HandoffDecision: + """Determine if a subtask should be handed off to another agent. + + Args: + subtask: Subtask definition with complexity, capabilities, etc. + agent_id: Current agent ID + agent_capabilities: Capabilities advertised by the current agent + + Returns: + HandoffDecision: Decision with reasoning + """ + complexity = subtask.get("complexity", "medium") + required_capabilities = subtask.get("required_capabilities", ["general"]) + estimated_time = subtask.get("estimated_time", 30) + current_capabilities = _expanded_capabilities(agent_capabilities or ["general"]) + + # Decision criteria: + # 1. High complexity tasks should be handed off to specialist agents + # 2. Tasks requiring specialized capabilities should go to specialists + # 3. Tasks estimated to take > 60 minutes should be broken down further + + # Check complexity threshold + if complexity == "high": + logger.info(f"High complexity subtask detected: {subtask['description']}") + return HandoffDecision( + should_handoff=True, + reason="Task complexity exceeds agent capability threshold", + target_capabilities=required_capabilities + ) + + # Check for specialized capabilities + specialized_capabilities = [ + cap for cap in required_capabilities + if cap not in current_capabilities + ] + + if specialized_capabilities: + logger.info(f"Specialized capabilities required: {specialized_capabilities}") + return HandoffDecision( + should_handoff=True, + reason=f"Requires specialized capabilities: {', '.join(specialized_capabilities)}", + target_capabilities=specialized_capabilities + ) + + # Check estimated time + if estimated_time > 60: + logger.info(f"Long-running task detected: {estimated_time} minutes") + return HandoffDecision( + should_handoff=True, + reason=f"Task estimated to take {estimated_time} minutes (threshold: 60)", + target_capabilities=required_capabilities + ) + + # No handoff needed + return HandoffDecision( + should_handoff=False, + reason="Task within agent capability", + target_capabilities=[] + ) + + +def select_target_agent( + available_agents: List[dict], + required_capabilities: List[str] +) -> str: + """Select the best agent for a handoff based on capabilities. + + Args: + available_agents: List of available agent metadata + required_capabilities: Required capabilities for the task + + Returns: + str: Selected agent ID or None if no suitable agent found + """ + # Score each agent based on capability match + best_agent = None + best_score = -1 + + for agent in available_agents: + agent_capabilities = set(agent.get("capabilities", [])) + required_set = set(required_capabilities) + + # Calculate match score + matches = len(agent_capabilities.intersection(required_set)) + total_required = len(required_set) + + if total_required > 0: + score = matches / total_required + else: + score = 0 + + # Prefer agents with exact capability match + if score > best_score: + best_score = score + best_agent = agent + + if best_agent: + logger.info( + f"Selected agent {best_agent['agent_id']} " + f"with score {best_score:.2f} for capabilities {required_capabilities}" + ) + return best_agent["agent_id"] + + logger.warning(f"No suitable agent found for capabilities: {required_capabilities}") + return None + + +def estimate_task_complexity(description: str) -> str: + """Estimate task complexity based on description keywords. + + Args: + description: Task description + + Returns: + str: "low", "medium", or "high" + """ + description_lower = description.lower() + + # High complexity indicators + high_complexity_keywords = [ + "refactor", "redesign", "architecture", "migrate", + "optimize", "performance", "security", "scale", + "distributed", "concurrent", "async", "parallel" + ] + + # Low complexity indicators + low_complexity_keywords = [ + "fix typo", "update comment", "rename", "format", + "add log", "simple", "trivial", "quick" + ] + + # Check for high complexity + if any(keyword in description_lower for keyword in high_complexity_keywords): + return "high" + + # Check for low complexity + if any(keyword in description_lower for keyword in low_complexity_keywords): + return "low" + + # Default to medium + return "medium" diff --git a/agent/main.py b/agent/main.py new file mode 100644 index 0000000..a89084a --- /dev/null +++ b/agent/main.py @@ -0,0 +1,629 @@ +"""Agent main loop with WebSocket connection to orchestrator. + +Merged from agent_swarm_v4: +- Bounded concurrency, duplicate protection, capacity reporting (available_slots) +- Serialized websocket sends (safe_send) and reconnect-with-reregister logic +- Per-task timeout and graceful task cancellation +- Separate repository root from per-task execution workspace (see git_operations) +- Peer collaboration: request/reply routing fixed relative to v4 (peer_waiters is now + initialized and inbound peer replies are routed back to the waiting coroutine; inbound + peer queries are answered with a lightweight, cost-free acknowledgement) +- OpenAI-only task executor (see task_executor) +- Preserves heicode handoff wiring, Manager-facing metrics, and env-var entrypoint +""" +import asyncio +import json +import logging +import os +import signal +import time +import uuid +from pathlib import Path +from typing import Optional + +import websockets +from dotenv import load_dotenv +from pydantic import BaseModel, Field, ValidationError +from prometheus_client import Counter, Gauge, Histogram, start_http_server +from websockets.exceptions import ConnectionClosed + +from .git_operations import GitOperations +from .task_executor import TaskExecutor + + +load_dotenv() + + +# load_dotenv() above reads a local .env (gitignored) so credentials need not be passed on +# the command line. These defaults just guarantee the vars exist; an empty OPENAI_API_KEY is +# still falsy, so TaskExecutor raises a clear error rather than silently using no key. +# Real deployments supply the key via the environment / secret_ref, NOT a committed file. +os.environ.setdefault("OPENAI_API_KEY", "") +os.environ.setdefault("OPENAI_API_BASE", "https://api.openai.com/v1") +os.environ.setdefault("OPENAI_MODEL", "gpt-4o-mini") + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) + +TASKS_EXECUTED = Counter("agent_tasks_executed_total", "Total tasks executed") +TASKS_FAILED = Counter("agent_tasks_failed_total", "Total tasks failed") +TASK_DURATION = Histogram("agent_task_duration_seconds", "Task execution duration") +HANDOFFS_INITIATED = Counter("agent_handoffs_initiated_total", "Total handoffs initiated") +AGENT_STATUS = Gauge("agent_status", "Agent status (0=idle, 1=busy, 2=failed)") +WEBSOCKET_RECONNECTS = Counter("agent_websocket_reconnects_total", "WebSocket reconnection attempts") +TASKS_REJECTED = Counter("agent_tasks_rejected_total", "Total tasks rejected due to capacity") +TASKS_DUPLICATE = Counter("agent_tasks_duplicate_total", "Total duplicate task assignments") +ACTIVE_TASKS = Gauge("agent_active_tasks", "Current number of active tasks") + + +class TaskAssignment(BaseModel): + task_id: str + description: str + context: dict = Field(default_factory=dict) + + +class AgentRuntimeDisconnected(RuntimeError): + """Raised when the runtime attempts to send without an active websocket.""" + + +class Agent: + """Agent that connects to orchestrator and executes tasks.""" + + MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4")) + TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60")) + HEARTBEAT_INTERVAL_SECONDS = 15 + + def __init__( + self, + orchestrator_url: str, + agent_id: Optional[str] = None, + capabilities: Optional[list[str]] = None, + workspace_dir: str = "/workspace", + git_repo_url: Optional[str] = None, + ): + # Initialize connection state, execution limits, and workspace helpers for this agent runtime. + self.agent_id = agent_id or f"agent-{uuid.uuid4().hex[:8]}" + self.orchestrator_url = orchestrator_url.rstrip("/") + self.capabilities = capabilities or ["general"] + self.workspace_dir = Path(workspace_dir) + self.git_repo_url = git_repo_url + + self.websocket: Optional[websockets.WebSocketClientProtocol] = None + self.running = False + self.current_task_id: Optional[str] = None + self.send_lock = asyncio.Lock() + self.task_semaphore = asyncio.Semaphore(self.MAX_CONCURRENT_TASKS) + self.active_tasks: dict[str, asyncio.Task] = {} + self.heartbeat_task: Optional[asyncio.Task] = None + # Outstanding peer-collaboration requests awaiting a reply, keyed by correlation_id. + self.peer_waiters: dict[str, asyncio.Future] = {} + # Summary of this agent's most recently completed task, shared when peers consult it. + self.last_summary: Optional[str] = None + + self.workspace_git = GitOperations(str(self.workspace_dir), self.agent_id) + + def available_slots(self) -> int: + # Return how many additional tasks this agent can currently accept. + return max(0, self.MAX_CONCURRENT_TASKS - len(self.active_tasks)) + + def task_workspace(self, task_id: str) -> Path: + # Compute the dedicated per-task working directory inside the agent workspace. + return self.workspace_dir / ".agent_tasks" / task_id + + async def safe_send(self, payload: dict): + # Serialize and send a websocket message while holding a lock to prevent concurrent writes. + if not self.websocket: + raise AgentRuntimeDisconnected("websocket unavailable") + async with self.send_lock: + await self.websocket.send(json.dumps(payload)) + + async def connect(self) -> bool: + # Open a websocket connection to the orchestrator and cache the live socket on success. + try: + self.websocket = await websockets.connect(f"{self.orchestrator_url}/ws/{self.agent_id}") + logger.info(f"Connected to orchestrator at {self.orchestrator_url}") + return True + except Exception as e: + logger.error(f"Failed to connect to orchestrator: {e}") + return False + + async def connect_with_retry(self) -> bool: + # Repeatedly attempt to connect with exponential backoff until connected or shutting down. + delay = 1 + while self.running: + if await self.connect(): + return True + WEBSOCKET_RECONNECTS.inc() + await asyncio.sleep(min(delay, 60)) + delay = min(delay * 2, 60) + return False + + async def register(self): + # Announce this agent and its current capacity to the orchestrator after connecting. + try: + await self.safe_send({ + "type": "register", + "agent_id": self.agent_id, + "capabilities": self.capabilities, + # Backward-compatible extra fields; older orchestrators ignore them. + "available_slots": self.available_slots(), + "active_task_ids": list(self.active_tasks.keys()), + }) + logger.info(f"Registered agent {self.agent_id} with capabilities: {self.capabilities}") + return True + except Exception as e: + logger.error(f"Failed to register: {e}") + return False + + async def send_heartbeat(self): + # Send a periodic liveness and capacity update to the orchestrator. + await self.safe_send({ + "type": "heartbeat", + "agent_id": self.agent_id, + "timestamp": time.time(), + "active_tasks": len(self.active_tasks), + "available_slots": self.available_slots(), + }) + + async def heartbeat_loop(self): + # Keep sending heartbeat messages until the connection closes or the agent stops running. + while self.running: + try: + await self.send_heartbeat() + except (ConnectionClosed, AgentRuntimeDisconnected): + return + except Exception as e: + logger.error(f"Failed to send heartbeat: {e}") + return + await asyncio.sleep(self.HEARTBEAT_INTERVAL_SECONDS) + + async def send_status_update(self, status: str, task_id: Optional[str] = None, message: str = ""): + # Report a human-readable task or agent status transition to the orchestrator. + try: + await self.safe_send({ + "type": "status_update", + "agent_id": self.agent_id, + "status": status, + "task_id": task_id, + "message": message, + "timestamp": time.time(), + }) + except Exception as e: + logger.error(f"Failed to send status update: {e}") + + async def request_peer_collaboration( + self, + task_id: str, + target_agent_id: str, + content: str, + timeout_seconds: float = 20.0, + ): + # Ask a peer agent (via the orchestrator) for guidance and wait for its reply. + correlation_id = f"peer-{task_id}-{uuid.uuid4().hex[:8]}" + loop = asyncio.get_running_loop() + waiter = loop.create_future() + self.peer_waiters[correlation_id] = waiter + try: + await self.safe_send({ + "type": "peer_message", + "agent_id": self.agent_id, + "target_agent_id": target_agent_id, + "task_id": task_id, + "content": content, + "correlation_id": correlation_id, + "is_reply": False, + "timestamp": time.time(), + }) + return await asyncio.wait_for(waiter, timeout=timeout_seconds) + finally: + self.peer_waiters.pop(correlation_id, None) + + @staticmethod + def _summarize_result(result: dict) -> Optional[str]: + # Extract a short, human-readable summary from an execution result. + if not isinstance(result, dict): + return None + for subtask_result in reversed(result.get("subtasks", []) or []): + summary = subtask_result.get("summary") + if isinstance(summary, str) and summary.strip(): + return summary.strip() + summary = result.get("summary") + return summary.strip() if isinstance(summary, str) and summary.strip() else None + + async def answer_peer_query(self, message: dict): + # Respond to an inbound peer query, sharing what this agent has actually done so far. + requester = message.get("from_agent_id") or message.get("agent_id") + correlation_id = message.get("correlation_id") + if not requester or not correlation_id: + return + shared = ( + f"My latest result: {self.last_summary}" + if self.last_summary + else "I have no completed result to share yet." + ) + reply_content = ( + f"From {self.agent_id} (capabilities: {', '.join(self.capabilities)}). {shared} " + "Treat implementation artifacts as the source of truth for behavior and exception semantics." + ) + try: + await self.safe_send({ + "type": "peer_message", + "agent_id": self.agent_id, + "target_agent_id": requester, + "task_id": message.get("task_id"), + "content": reply_content, + "correlation_id": correlation_id, + "is_reply": True, + "timestamp": time.time(), + }) + except Exception as e: + logger.error(f"Failed to answer peer query: {e}") + + async def send_task_result(self, task_id: str, success: bool, result: dict): + # Send a completion or failure payload for a finished task execution. + try: + payload = { + "type": "task_complete" if success else "task_failed", + "agent_id": self.agent_id, + "task_id": task_id, + "timestamp": time.time(), + } + if success: + payload["result"] = result + else: + payload["reason"] = result.get("error", "Task failed") + payload["result"] = result + await self.safe_send(payload) + except Exception as e: + logger.error(f"Failed to send task result: {e}") + + async def send_blocked_on_handoff(self, task_id: str, result: dict): + # Report that a task is paused because execution delegated work to a child handoff task. + try: + child_task_id = None + for subtask_result in result.get("subtasks", []): + if subtask_result.get("status") == "handed_off": + child_task_id = subtask_result.get("child_task_id") + break + await self.safe_send({ + "type": "blocked_on_handoff", + "agent_id": self.agent_id, + "task_id": task_id, + "child_task_id": child_task_id, + "reason": "Waiting on delegated child task", + "result": result, + "timestamp": time.time(), + }) + except Exception as e: + logger.error(f"Failed to send blocked_on_handoff: {e}") + + async def request_handoff(self, task_id: str, subtask: dict, target_capabilities: list[str]): + # Ask the orchestrator to delegate a discovered subtask to another capable agent. + try: + await self.safe_send({ + "type": "handoff_request", + "agent_id": self.agent_id, + "task_id": task_id, + "subtask": subtask, + "target_capabilities": target_capabilities, + "timestamp": time.time(), + }) + HANDOFFS_INITIATED.inc() + except Exception as e: + logger.error(f"Failed to request handoff: {e}") + + async def execute_assignment(self, assignment: TaskAssignment): + # Execute one accepted assignment, manage workspace/git flow, and publish lifecycle updates. + async with self.task_semaphore: + task_id = assignment.task_id + description = assignment.description + context = assignment.context or {} + self.current_task_id = task_id + ACTIVE_TASKS.set(len(self.active_tasks)) + AGENT_STATUS.set(1) + + await self.safe_send({ + "type": "task_start", + "agent_id": self.agent_id, + "task_id": task_id, + "timestamp": time.time(), + }) + await self.send_status_update("busy", task_id, "Starting task execution") + + start_time = time.time() + task_workspace = self.task_workspace(task_id) + task_workspace.mkdir(parents=True, exist_ok=True) + + git_enabled = False + try: + executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(task_workspace)) + git_enabled = await self.workspace_git.is_git_workspace() + if git_enabled: + branch_created = await self.workspace_git.create_result_branch(task_id) + if not branch_created: + logger.warning("Failed to create task result branch; continuing without git push") + git_enabled = False + + result = await asyncio.wait_for( + executor.execute_task( + task_id=task_id, + description=description, + context={ + **context, + "workspace_dir": str(task_workspace), + "repo_workspace_dir": str(self.workspace_dir), + "git_repo_url": self.git_repo_url, + "agent_id": self.agent_id, + }, + handoff_callback=self.request_handoff, + agent_capabilities=self.capabilities, + peer_collaboration_callback=self.request_peer_collaboration, + ), + timeout=self.TASK_TIMEOUT_SECONDS, + ) + + awaiting_handoff = result.get("awaiting_handoff", False) + if result.get("success"): + self.last_summary = self._summarize_result(result) or self.last_summary + if result.get("success") and not awaiting_handoff: + if git_enabled: + commit_sha = await self.workspace_git.commit_changes( + message=f"Task {task_id}: {description[:50]}" + ) + if commit_sha: + branch_name = await self.workspace_git.push_results() + result["git_branch"] = branch_name + result["commit_sha"] = commit_sha + else: + result["git_skipped"] = "No workspace changes to commit" + else: + result["git_skipped"] = "Workspace is not a Git checkout" + + if awaiting_handoff: + await self.send_blocked_on_handoff(task_id, result) + else: + await self.send_task_result(task_id, result.get("success", False), result) + + duration = time.time() - start_time + TASK_DURATION.observe(duration) + if result.get("success"): + TASKS_EXECUTED.inc() + else: + TASKS_FAILED.inc() + + AGENT_STATUS.set(0) + if awaiting_handoff: + await self.send_status_update("handoff-pending", task_id, "Waiting for delegated child task") + await self.send_status_update("idle", None, "Delegated child task created") + else: + await self.send_status_update("idle", None, "Task completed") + + except asyncio.TimeoutError: + TASKS_FAILED.inc() + AGENT_STATUS.set(2) + await self.send_task_result(task_id, False, {"error": "timeout", "success": False}) + await self.send_status_update("idle", None, "Task failed: timeout") + except asyncio.CancelledError: + await self.send_task_result(task_id, False, {"error": "cancelled", "success": False}) + await self.send_status_update("idle", None, "Task cancelled") + raise + except Exception as e: + logger.error(f"Error executing task {task_id}: {e}") + TASKS_FAILED.inc() + AGENT_STATUS.set(2) + await self.send_task_result(task_id, False, {"error": str(e), "success": False}) + await self.send_status_update("idle", None, f"Task failed: {e}") + finally: + self.active_tasks.pop(task_id, None) + ACTIVE_TASKS.set(len(self.active_tasks)) + self.current_task_id = None + if not self.active_tasks: + AGENT_STATUS.set(0) + + async def cancel_task(self, task_id: str): + # Cancel an actively running asyncio task if the orchestrator requests termination. + task = self.active_tasks.get(task_id) + if task: + task.cancel() + + async def handle_task_assignment(self, message: dict): + # Validate an incoming assignment, reject duplicates/capacity overflow, and start execution. + try: + assignment = TaskAssignment(**message) + except ValidationError: + logger.warning(f"invalid task assignment: {message}") + return + + if assignment.task_id in self.active_tasks: + TASKS_DUPLICATE.inc() + await self.safe_send({ + "type": "task_accepted", + "task_id": assignment.task_id, + "status": "duplicate", + }) + return + + if len(self.active_tasks) >= self.MAX_CONCURRENT_TASKS: + TASKS_REJECTED.inc() + await self.safe_send({ + "type": "task_rejected", + "task_id": assignment.task_id, + "reason": "at_capacity", + "available_slots": self.available_slots(), + }) + return + + await self.safe_send({ + "type": "task_accepted", + "task_id": assignment.task_id, + "available_slots": self.available_slots() - 1, + }) + + task = asyncio.create_task(self.execute_assignment(assignment)) + self.active_tasks[assignment.task_id] = task + ACTIVE_TASKS.set(len(self.active_tasks)) + + def _resolve_peer_reply(self, message: dict) -> bool: + # Resolve the waiting future for an inbound peer reply; return True if it was a reply. + correlation_id = message.get("correlation_id") + if not correlation_id: + return False + waiter = self.peer_waiters.get(correlation_id) + if waiter and not waiter.done(): + waiter.set_result(message) + return True + # A correlation we own but already resolved/timed out: treat as handled reply. + return bool(message.get("is_reply")) + + async def handle_peer_message(self, message: dict): + # Route a peer message: resolve our own pending request, or answer an inbound query. + if message.get("is_reply") or message.get("correlation_id") in self.peer_waiters: + handled = self._resolve_peer_reply(message) + if handled: + return + logger.info( + "Received peer query for task %s from %s", + message.get("task_id"), + message.get("from_agent_id") or message.get("agent_id"), + ) + await self.answer_peer_query(message) + + async def handle_message(self, message: dict): + # Route each inbound orchestrator message to the appropriate handler. + msg_type = message.get("type") + control_messages = {"registered", "heartbeat_ack", "task_completed", "task_failed_ack", "task_blocked_ack"} + + if msg_type == "task_assignment": + await self.handle_task_assignment(message) + elif msg_type == "handoff_response": + logger.info(f"Handoff accepted for task {message.get('task_id')}") + elif msg_type == "cancel_task": + await self.cancel_task(message["task_id"]) + elif msg_type == "ping": + await self.safe_send({"type": "pong"}) + elif msg_type == "peer_message": + await self.handle_peer_message(message) + elif msg_type in control_messages: + logger.debug(f"Received control message: {msg_type}") + else: + logger.warning(f"Unknown message type: {msg_type}") + + async def message_loop(self): + # Continuously receive websocket messages and dispatch them until the connection ends. + async for raw in self.websocket: + try: + data = json.loads(raw) + await self.handle_message(data) + except json.JSONDecodeError as e: + logger.error(f"Failed to parse message: {e}") + except Exception as e: + logger.error(f"Error handling message: {e}") + + async def shutdown_active_tasks(self): + # Cancel and await all active task coroutines during agent shutdown. + tasks = list(self.active_tasks.values()) + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + def install_signal_handlers(self): + # Register process signal handlers that trigger a graceful runtime shutdown. + def _handle_shutdown(): + # Flip the runtime into shutdown mode and close the websocket asynchronously. + logger.info("shutdown signal received") + self.running = False + if self.websocket: + asyncio.create_task(self.websocket.close()) + + for sig in (signal.SIGINT, signal.SIGTERM): + try: + asyncio.get_running_loop().add_signal_handler(sig, _handle_shutdown) + except NotImplementedError: + # add_signal_handler is unsupported on Windows event loops; skip gracefully. + pass + + async def run(self): + # Run the full agent lifecycle: startup, registration, message processing, reconnect, and cleanup. + self.running = True + self.install_signal_handlers() + + if os.getenv("METRICS_PORT"): + start_http_server(int(os.getenv("METRICS_PORT", "9000"))) + + if self.git_repo_url: + logger.info(f"Cloning workspace from {self.git_repo_url}") + if not await self.workspace_git.clone_workspace(self.git_repo_url): + logger.error("Failed to clone workspace, exiting") + return + + while self.running: + connected = await self.connect_with_retry() + if not connected: + break + + if not await self.register(): + logger.error("Failed to register with orchestrator, exiting") + return + + self.heartbeat_task = asyncio.create_task(self.heartbeat_loop()) + try: + await self.message_loop() + except ConnectionClosed: + logger.warning("connection closed; reconnecting") + except Exception as e: + logger.error(f"Error in message loop: {e}") + finally: + if self.heartbeat_task: + self.heartbeat_task.cancel() + try: + await self.heartbeat_task + except asyncio.CancelledError: + pass + self.heartbeat_task = None + + if self.websocket: + try: + await self.safe_send({"type": "deregister", "agent_id": self.agent_id}) + except Exception: + pass + try: + await self.websocket.close() + except Exception: + pass + self.websocket = None + + if self.running: + await asyncio.sleep(1) + + await self.shutdown_active_tasks() + logger.info("Agent shutdown complete") + + +async def main(): + # Build an agent instance from environment configuration and start its runtime loop. + orchestrator_url = os.getenv("ORCHESTRATOR_URL", "ws://localhost:8000") + agent_id = os.getenv("AGENT_ID") + capabilities = os.getenv("AGENT_CAPABILITIES", "general").split(",") + workspace_dir = os.getenv("WORKSPACE_DIR", "/workspace") + git_repo_url = os.getenv("GIT_REPO_URL") + + logger.info(f"Starting agent with ID: {agent_id or 'auto-generated'}") + logger.info(f"Capabilities: {capabilities}") + logger.info(f"Orchestrator URL: {orchestrator_url}") + + agent = Agent( + orchestrator_url=orchestrator_url, + agent_id=agent_id, + capabilities=capabilities, + workspace_dir=workspace_dir, + git_repo_url=git_repo_url, + ) + await agent.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/agent/requirements.txt b/agent/requirements.txt new file mode 100644 index 0000000..88645b0 --- /dev/null +++ b/agent/requirements.txt @@ -0,0 +1,5 @@ +openai==1.55.3 +websockets==13.1 +pydantic==2.9.2 +python-dotenv==1.0.1 +prometheus-client==0.20.0 diff --git a/agent/task_executor.py b/agent/task_executor.py new file mode 100644 index 0000000..c32178c --- /dev/null +++ b/agent/task_executor.py @@ -0,0 +1,480 @@ +"""Task execution engine with OpenAI-compatible LLM provider integration. + +Merged from agent_swarm_v4: +- OpenAI-only provider configuration (OPENAI_*/MODEL_* env vars and custom base URLs) +- Peer-collaboration hook and specialist-role alignment prompting +- Preserves model invocation, handoff decision hooks, workspace summarization, + file-application behavior, and the Manager billing/audit usage attribution + (X-Agent/X-Agnet headers, usage payload, billing_source) +""" +import asyncio +import json +import logging +import os +import time +from pathlib import Path +from typing import Callable, Optional + +from openai import AsyncOpenAI + +from .handoff_logic import should_handoff + +logger = logging.getLogger(__name__) + + +class TaskExecutor: + """Executes tasks using a configurable OpenAI-compatible API.""" + + def __init__(self, agent_id: str, workspace_dir: str): + self.agent_id = agent_id + self.workspace_dir = workspace_dir + + api_key = ( + os.getenv("OPENAI_API_KEY") + or os.getenv("MODEL_API_KEY") + ) + if not api_key: + raise ValueError("OPENAI_API_KEY environment variable not set") + + self.model = ( + os.getenv("OPENAI_MODEL") + or os.getenv("MODEL_NAME") + or os.getenv("MODEL_ID") + or "gpt-4o-mini" + ) + self.api_base = ( + os.getenv("OPENAI_API_BASE") + or os.getenv("MODEL_API_BASE") + or "https://api.openai.com/v1" + ) + self.api_mode = "openai" + self.client = AsyncOpenAI(api_key=api_key, base_url=self.api_base) + self.client_type = "openai" + + self.usage = self._empty_usage() + self.current_context: dict = {} + + async def execute_task( + self, + task_id: str, + description: str, + context: dict, + handoff_callback: Optional[Callable] = None, + agent_capabilities: Optional[list[str]] = None, + peer_collaboration_callback: Optional[Callable] = None, + ) -> dict: + logger.info(f"Executing task {task_id}: {description}") + + try: + started_at = time.time() + self.current_context = context or {} + self.usage = self._empty_usage(context) + multi_agent_leaf_mode = self._multi_agent_leaf_mode(context) + allow_handoff = self._allow_dynamic_handoff(context) + + if multi_agent_leaf_mode: + subtasks = [{ + "description": description, + "complexity": "medium", + "estimated_time": 30, + "dependencies": [], + "required_capabilities": context.get("required_capabilities") or ["general"], + }] + elif self._subtask_handoff_enabled(): + subtasks = await self._parse_task(description, context) + else: + subtasks = [{ + "description": description, + "complexity": "medium", + "estimated_time": 30, + "dependencies": [], + "required_capabilities": ["general"], + }] + + results = [] + for subtask in subtasks: + if handoff_callback and self._subtask_handoff_enabled() and allow_handoff: + decision = should_handoff( + subtask, + self.agent_id, + agent_capabilities=agent_capabilities, + ) + if decision.should_handoff: + await handoff_callback( + task_id=task_id, + subtask=subtask, + target_capabilities=decision.target_capabilities, + ) + results.append({ + "subtask": subtask, + "status": "handed_off", + "reason": decision.reason, + "target_capabilities": decision.target_capabilities, + }) + continue + + results.append(await self._execute_subtask(subtask, task_id, context, peer_collaboration_callback)) + + success = all(r["status"] in ["completed", "handed_off"] for r in results) + awaiting_handoff = any(r["status"] == "handed_off" for r in results) + return { + "success": success, + "task_id": task_id, + "subtasks": results, + "awaiting_handoff": awaiting_handoff, + "agent_id": self.agent_id, + "usage": self._usage_payload(time.time() - started_at), + } + except Exception as e: + logger.error(f"Error executing task {task_id}: {e}") + return { + "success": False, + "task_id": task_id, + "error": str(e), + "agent_id": self.agent_id, + "usage": self._usage_payload(time.time() - started_at if "started_at" in locals() else 0), + } + finally: + self.current_context = {} + + async def _parse_task(self, description: str, context: dict) -> list[dict]: + try: + prompt = f"""You are a task planning assistant. Break down the following programming task into concrete, actionable subtasks. + +Task: {description} + +Context: {json.dumps(context, indent=2)} + +Return a JSON array of subtasks, where each subtask has: +- description: Clear description of what needs to be done +- complexity: \"low\", \"medium\", or \"high\" +- estimated_time: Estimated time in minutes +- dependencies: List of subtask indices this depends on (empty if none) +- required_capabilities: List of capabilities needed + +Return ONLY the JSON array, no other text.""" + content = await self._complete(prompt, max_tokens=2000) + return json.loads(self._strip_json_fence(content)) + except Exception as e: + logger.error(f"Error parsing task: {e}") + return [{ + "description": description, + "complexity": "medium", + "estimated_time": 30, + "dependencies": [], + "required_capabilities": ["general"], + }] + + async def _execute_subtask(self, subtask: dict, task_id: str, context: dict, peer_collaboration_callback: Optional[Callable]) -> dict: + try: + description = subtask["description"] + workspace_files = self._summarize_workspace() + workspace_context = self._collect_workspace_context() + user_prompt = context.get("user_prompt") or context.get("run_goal") or context.get("root_task_description") or "" + specialist_role = context.get("specialist_role", "general") + dependency_artifacts = context.get("dependency_artifacts") or [] + implementation_artifacts = [ + artifact for artifact in dependency_artifacts + if "implementation" in (artifact.get("task_id") or "") + ] + testing_artifacts = [ + artifact for artifact in dependency_artifacts + if "testing" in (artifact.get("task_id") or "") + ] + peer_context = [] + peer_agents = context.get("peer_agents") or [] + should_consult_peers = specialist_role in {"testing", "documentation"} + if peer_collaboration_callback and peer_agents and should_consult_peers: + max_peer_consults = int((context or {}).get("max_peer_consults", 2) or 2) + preferred_roles = [] + if specialist_role in {"testing", "documentation"}: + preferred_roles = ["implementation"] + + ordered_peers = sorted( + peer_agents, + key=lambda peer: 0 if peer.get("role") in preferred_roles else 1, + ) + + for peer in ordered_peers[:max_peer_consults]: + try: + reply = await peer_collaboration_callback( + task_id=task_id, + target_agent_id=peer.get("agent_id"), + content=( + f"Specialist role: {specialist_role}. " + f"Please provide guidance and confirm behavior for: {description}. " + f"Original user request: {user_prompt}" + ), + timeout_seconds=float((context or {}).get("peer_timeout_seconds", 10.0) or 10.0), + ) + peer_context.append( + { + "agent_id": peer.get("agent_id"), + "role": peer.get("role"), + "content": reply.get("content", ""), + } + ) + except Exception as exc: + logger.warning(f"Peer collaboration failed with {peer.get('agent_id')}: {exc}") + prompt = f"""You are a programming assistant working in a collaborative agent system. + +Original user request: +{user_prompt} + +Your specialist role: +{specialist_role} + +Task: {description} + +Dependency artifacts from other specialists: +{json.dumps(dependency_artifacts, indent=2)} + +Implementation artifacts relevant to alignment: +{json.dumps(implementation_artifacts, indent=2)} + +Testing artifacts relevant to alignment: +{json.dumps(testing_artifacts, indent=2)} + +Workspace directory: {self.workspace_dir} +Current workspace files: +{json.dumps(workspace_files, indent=2)} +Relevant workspace file contents: +{json.dumps(workspace_context, indent=2)} +Peer specialist input: +{json.dumps(peer_context, indent=2)} + +Alignment requirements: +- If your role is testing, align your tests with the implementation artifacts and their stated error semantics. +- If your role is documentation, align your docs with both implementation and testing artifacts. +- Do not invent behavior that conflicts with dependency artifacts unless you explicitly surface an error. +- Treat implementation artifacts as the source of truth for API behavior and exception semantics. +- If peer specialist input conflicts with implementation artifacts, prefer implementation semantics and explain the correction in your changes summary. +- If your role is documentation or testing, update only your specialist outputs to converge on implementation behavior unless the implementation artifact is clearly missing or contradictory. + +Return your response as JSON with this structure: +{{ + \"status\": \"completed\" or \"failed\", + \"summary\": \"Brief description of what was done\", + \"files\": [{{\"path\": \"relative/path.py\", \"action\": \"write\", \"content\": \"complete file content\"}}], + \"changes\": \"Detailed description of changes\", + \"error\": \"Error message if failed, null otherwise\" +}} + +Return ONLY the JSON, no other text.""" + content = await self._complete(prompt, max_tokens=4000) + result = self._parse_json_response(content) + if result.get("status") == "completed": + apply_result = await self._apply_file_changes(result.get("files", [])) + result["files_modified"] = apply_result["files_modified"] + result["files_deleted"] = apply_result["files_deleted"] + if apply_result["errors"]: + result["status"] = "failed" + result["error"] = "; ".join(apply_result["errors"]) + result["subtask"] = subtask + return result + except Exception as e: + logger.error(f"Error executing subtask: {e}") + return { + "subtask": subtask, + "status": "failed", + "error": str(e), + "summary": f"Failed to execute: {e}", + } + + async def _complete(self, prompt: str, max_tokens: int) -> str: + extra_headers = self._model_attribution_headers() + response = await self.client.chat.completions.create( + model=self.model, + messages=[{"role": "user", "content": prompt}], + max_tokens=max_tokens, + extra_headers=extra_headers or None, + ) + self._record_openai_usage(response) + return response.choices[0].message.content or "" + + def _empty_usage(self, context: Optional[dict] = None) -> dict: + plan = ((context or {}).get("orchestration_plan") or {}) + billing = plan.get("billing_context") or {} + return { + "model_id": self.model, + "model_tokens": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "model_cost_usd": 0.0, + "runtime_seconds": 0.0, + "billing_source": billing.get("provider") or os.getenv("BILLING_SOURCE", "unknown"), + } + + def _record_openai_usage(self, response): + usage = getattr(response, "usage", None) + if not usage: + return + prompt_tokens = int(getattr(usage, "prompt_tokens", 0) or 0) + completion_tokens = int(getattr(usage, "completion_tokens", 0) or 0) + total_tokens = int(getattr(usage, "total_tokens", 0) or prompt_tokens + completion_tokens) + self._add_usage(prompt_tokens, completion_tokens, total_tokens) + + def _add_usage(self, prompt_tokens: int, completion_tokens: int, total_tokens: int): + self.usage["prompt_tokens"] += prompt_tokens + self.usage["completion_tokens"] += completion_tokens + self.usage["model_tokens"] += total_tokens + input_cost = float(os.getenv("MODEL_INPUT_COST_PER_1M", "0") or 0) + output_cost = float(os.getenv("MODEL_OUTPUT_COST_PER_1M", "0") or 0) + self.usage["model_cost_usd"] += ( + prompt_tokens * input_cost / 1_000_000 + + completion_tokens * output_cost / 1_000_000 + ) + + def _usage_payload(self, runtime_seconds: float) -> dict: + usage = dict(self.usage) + usage["runtime_seconds"] = runtime_seconds + return usage + + def _model_attribution_headers(self) -> dict: + headers = {} + mapping = { + "manager_deployment_id": ["X-Agent-Manager-Deployment-ID", "X-Agnet-Manager-Deployment-ID"], + "swarm_id": ["X-Agent-Swarm-ID", "X-Agnet-Swarm-ID"], + "task_id": ["X-Agent-Task-ID", "X-Agnet-Task-ID"], + "agent_role": ["X-Agent-Agent-Role", "X-Agnet-Agent-Role"], + "correlation_id": ["X-Correlation-ID"], + "model_id": ["X-Agent-Model-ID", "X-Agnet-Model-ID"], + } + for key, header_names in mapping.items(): + value = self.current_context.get(key) + if value: + for header in header_names: + headers[header] = str(value) + headers.setdefault("X-Agent-Model-ID", self.model) + headers.setdefault("X-Agnet-Model-ID", self.model) + return headers + + def _subtask_handoff_enabled(self) -> bool: + return os.getenv("ENABLE_SUBTASK_HANDOFF", "false").lower() in {"1", "true", "yes"} + + def _multi_agent_leaf_mode(self, context: Optional[dict]) -> bool: + return self._subtask_handoff_enabled() and (context or {}).get("workflow_mode") == "multi_agent" + + def _allow_dynamic_handoff(self, context: Optional[dict]) -> bool: + if (context or {}).get("workflow_mode") != "multi_agent": + return True + return bool((context or {}).get("allow_handoff", False)) + + def _strip_json_fence(self, content: str) -> str: + content = content.strip() + if not content.startswith("```"): + return content + lines = content.split("\n") + if lines and lines[0].startswith("```"): + lines = lines[1:] + if lines and lines[-1].strip() == "```": + lines = lines[:-1] + return "\n".join(lines).strip() + + def _parse_json_response(self, content: str) -> dict: + content = self._strip_json_fence(content) + try: + return json.loads(content) + except json.JSONDecodeError: + start = content.find("{") + end = content.rfind("}") + if start == -1 or end == -1 or end <= start: + raise + return json.loads(content[start:end + 1]) + + def _summarize_workspace(self, max_files: int = 80) -> list[str]: + root = Path(self.workspace_dir) + if not root.exists(): + return [] + ignored_dirs = {".git", "__pycache__", "node_modules", ".venv", "venv"} + files = [] + for path in root.rglob("*"): + if len(files) >= max_files: + break + if not path.is_file(): + continue + if any(part in ignored_dirs for part in path.relative_to(root).parts): + continue + files.append(str(path.relative_to(root))) + return sorted(files) + + def _collect_workspace_context(self, max_files: int = 20, max_bytes_per_file: int = 6000) -> dict[str, str]: + root = Path(self.workspace_dir) + if not root.exists(): + return {} + ignored_dirs = {".git", "__pycache__", "node_modules", ".venv", "venv"} + allowed_suffixes = {".py", ".js", ".ts", ".tsx", ".jsx", ".json", ".md", ".txt", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".sh"} + context = {} + for path in sorted(root.rglob("*")): + if len(context) >= max_files: + break + if not path.is_file(): + continue + relative = path.relative_to(root) + if any(part in ignored_dirs for part in relative.parts): + continue + if path.suffix and path.suffix.lower() not in allowed_suffixes: + continue + try: + data = path.read_bytes()[:max_bytes_per_file] + context[str(relative)] = data.decode("utf-8") + except UnicodeDecodeError: + continue + except Exception as e: + logger.warning(f"Failed to read workspace file {relative}: {e}") + return context + + def _resolve_workspace_path(self, file_path: str) -> Path: + if not file_path or os.path.isabs(file_path): + raise ValueError(f"Invalid relative path: {file_path}") + root = Path(self.workspace_dir).resolve() + resolved = (root / file_path).resolve() + if root != resolved and root not in resolved.parents: + raise ValueError(f"Path escapes workspace: {file_path}") + return resolved + + async def _apply_file_changes(self, files: list[dict]) -> dict: + result = {"files_modified": [], "files_deleted": [], "errors": []} + for file_change in files: + path = file_change.get("path") + action = file_change.get("action", "write") + try: + target = self._resolve_workspace_path(path) + if action == "write": + content = file_change.get("content") + if content is None: + raise ValueError(f"Missing content for {path}") + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + result["files_modified"].append(path) + elif action == "delete": + if target.exists(): + target.unlink() + result["files_deleted"].append(path) + else: + raise ValueError(f"Unsupported file action for {path}: {action}") + except Exception as e: + logger.error(f"Failed to apply file change {path}: {e}") + result["errors"].append(f"{path}: {e}") + return result + + async def read_file(self, file_path: str) -> Optional[str]: + try: + full_path = os.path.join(self.workspace_dir, file_path) + with open(full_path, "r", encoding="utf-8") as f: + return f.read() + except Exception as e: + logger.error(f"Error reading file {file_path}: {e}") + return None + + async def write_file(self, file_path: str, content: str) -> bool: + try: + full_path = self._resolve_workspace_path(file_path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "w", encoding="utf-8") as f: + f.write(content) + logger.info(f"Wrote file: {file_path}") + return True + except Exception as e: + logger.error(f"Error writing file {file_path}: {e}") + return False diff --git a/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.html b/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.html new file mode 100644 index 0000000..cf64621 --- /dev/null +++ b/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.html @@ -0,0 +1,115 @@ + + + + + + Swarm Terminal Snapshot + + + +
+
+
+
+ + + +
+
swarm-terminal-snapshot
+
swarm-300589052dd0
+
+
+
$ python3 scripts/run_swarm_poster_demo.py
+
[swarm] deployment_id=runtime-dep-90b592c531e3
+
[swarm] swarm_id=swarm-300589052dd0
+
[task] task_id=swarm-300589052dd0-task-1
+
[task] status=in_progress
+
[task] assigned_agent=agent-full-5d6968f886-7cm2l
+
+
$ curl -H "Authorization: Bearer ***" /api/swarms/{swarm_id}/tasks
+
task.started_at = 2026-05-29 10:12:36 UTC
+
task.halfway_at = 2026-05-29 10:12:56 UTC
+
halfway.observed_at = 2026-05-29 10:12:58 UTC
+
+
$ curl -H "Authorization: Bearer ***" /api/swarms/{swarm_id}/metrics
+
runtime.status = running
+
runtime.tasks_total = 1
+
runtime.tasks_by_status = {"in_progress": 1}
+
runtime.agents_connected = 1
+
runtime.budget.duration_seconds = 40
+
runtime.budget.duration_ratio = 495.7%
+
+
# milestone
+
> 任务开始: 2026-05-29 10:12:36 UTC
+
> 任务过半: 2026-05-29 10:12:58 UTC
+
+
+
+ + diff --git a/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.png b/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.png new file mode 100644 index 0000000..9cf986a Binary files /dev/null and b/artifacts/swarm-posters/swarm-poster-300589052dd0-terminal.png differ diff --git a/artifacts/swarm-posters/swarm-poster-300589052dd0.html b/artifacts/swarm-posters/swarm-poster-300589052dd0.html new file mode 100644 index 0000000..9faf500 --- /dev/null +++ b/artifacts/swarm-posters/swarm-poster-300589052dd0.html @@ -0,0 +1,285 @@ + + + + + + Swarm Blog MVP Live Demo + + + +
+
+
+
+
Live Swarm Poster
+

Swarm Blog MVP Live Demo

+

真实蜂群任务已成功创建，并已记录到“任务开始”和“进行过半”两段里程碑。下面的内容全部来自当前运行中的 swarm 状态，而不是手工拼接。

+
+
+ Demo Goal +

在目标仓库里开发一个博客系统 MVP，包括文章列表、详情和基础增删改能力。

+
+
+ +
+
+
+
+
Swarm
+
swarm-300589052dd0
+
+
+
Task Status
+
in_progress
+
+
+
Budget Progress
+
496%
+
+
+ + Timeline +
+
+

任务开始

+

任务已被 agent 领取并进入运行态。开始时间：2026-05-29 10:12:36 UTC

+
+
+

进行到一半

+

脚本按 budget 的 50% 自动确认里程碑。预算过半时间：2026-05-29 10:12:56 UTC，实际记录时间：2026-05-29 10:12:58 UTC

+
+
+

当前执行中

+

任务仍处于 in_progress，最近心跳：2026-05-29T10:12:49.319420Z

+
+
+
+ +
+
+
Poster Snapshot
+
Mid-Run
+
这张海报展示的是蜂群任务已经真正启动，并且已经跑到预算半程时的现场状态。
+
+ +
+ Swarm Meta +
Deploymentruntime-dep-90b592c531e3
+
Taskswarm-300589052dd0-task-1
+
Agentagent-full-5d6968f886-7cm2l
+
Attempts0
+
Budget40s
+
Output Fileswarm-poster-300589052dd0.html
+
+ +
+ Latest Events +
1task.running
+
2task.heartbeat
+
3budget.alert
+
注：如果后续继续执行，日志与状态还会变化；这张图锁定的是“开始后已过半”的那一刻。
+
+
+
+
+ + diff --git a/artifacts/swarm-posters/swarm-poster-300589052dd0.json b/artifacts/swarm-posters/swarm-poster-300589052dd0.json new file mode 100644 index 0000000..e6dc9ed --- /dev/null +++ b/artifacts/swarm-posters/swarm-poster-300589052dd0.json @@ -0,0 +1,94 @@ +{ + "swarm": { + "deployment_id": "runtime-dep-90b592c531e3", + "runtime_deployment_id": "runtime-dep-90b592c531e3", + "manager_deployment_id": "dep_poster_demo_20260529T100940Z", + "swarm_id": "swarm-300589052dd0", + "status": "running", + "created": true + }, + "task": { + "task_id": "swarm-300589052dd0-task-1", + "title": "Swarm objective", + "description": "Build a blog system MVP in the configured workspace repository. Deliver a runnable implementation with a simple backend and frontend for posts: list posts, view a post, create, edit, and delete. Keep setup minimal and include short run instructions.", + "status": "in_progress", + "agent_role": "general", + "required_capabilities": [ + "general" + ], + "depends_on": [], + "parent_task_id": null, + "root_task_id": "swarm-300589052dd0-task-1", + "source": "runtime_bridge", + "task_graph_id": "task-1", + "assigned_agent_id": "agent-full-5d6968f886-7cm2l", + "created_at": 1780049380.5639093, + "started_at": 1780049556.1651127, + "completed_at": null, + "attempt": 0, + "max_retries": 3, + "retry_count": 0, + "blocked_reason": null, + "context": { + "orchestration_plan": { + "objective": "Build a blog system MVP in the configured workspace repository. Deliver a runnable implementation with a simple backend and frontend for posts: list posts, view a post, create, edit, and delete. Keep setup minimal and include short run instructions.", + "sub_mode": "code", + "risk_level": "low", + "budget": { + "duration_seconds": 40, + "token_limit": 120000 + }, + "agents": [ + { + "task_id": "poster-blog-20260529t100940z", + "role": "fullstack", + "title": "Implement blog system MVP", + "description": "Create a runnable blog system MVP in the target repository with post list, post detail, and create/edit/delete flows. Keep the implementation pragmatic and coherent with the existing repo. Commit and push the result branch when changes are ready.", + "depends_on": [] + } + ] + }, + "resource_grants": [], + "sub_mode": "code", + "runtime_deployment_id": "runtime-dep-90b592c531e3", + "swarm_id": "swarm-300589052dd0", + "manager_deployment_id": "dep_poster_demo_20260529T100940Z", + "correlation_id": "corr_poster_demo_20260529T100940Z", + "task_graph_id": "task-1", + "task_title": "Swarm objective", + "depends_on": [], + "agent_role": "general", + "required_capabilities": [ + "general" + ], + "parent_task_id": null, + "root_task_id": "swarm-300589052dd0-task-1", + "source": "runtime_bridge", + "workflow_mode": "single_agent", + "allow_handoff": false, + "model_id": null + } + }, + "metrics": { + "swarm_id": "swarm-300589052dd0", + "window": "15m", + "step": "60s", + "status": "running", + "runtime_duration_seconds": 198.26700401306152, + "tasks_total": 1, + "tasks_by_status": { + "in_progress": 1 + }, + "agents_connected": 1, + "agents_registered": 1, + "average_task_duration_seconds": 0, + "budget": { + "duration_seconds": 40, + "duration_ratio": 4.956675100326538 + } + }, + "started_at": 1780049556.1651127, + "halfway_at": 1780049576.1651127, + "halfway_observed_at": 1780049578.828649, + "poster_html": "/Users/mac/Projects/HeiCode-Sawrm/artifacts/swarm-posters/swarm-poster-300589052dd0.html" +} \ No newline at end of file diff --git a/artifacts/swarm-posters/swarm-poster-300589052dd0.png b/artifacts/swarm-posters/swarm-poster-300589052dd0.png new file mode 100644 index 0000000..f19e284 Binary files /dev/null and b/artifacts/swarm-posters/swarm-poster-300589052dd0.png differ diff --git a/check_status.sh b/check_status.sh new file mode 100644 index 0000000..4ea71f0 --- /dev/null +++ b/check_status.sh @@ -0,0 +1,109 @@ +#!/bin/bash +# K8s 蜂群系统 - 快速状态检查 + +echo "======================================" +echo " K8s 蜂群系统 - 状态检查" +echo "======================================" +echo "" + +# 颜色定义 +GREEN='\033[0;32m' +RED='\033[0;31m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +# 检查 kubectl +if ! command -v kubectl &> /dev/null; then + echo -e "${RED}❌ kubectl 未安装${NC}" + exit 1 +fi + +# 检查集群连接 +echo "1. 检查集群连接..." +if kubectl cluster-info &> /dev/null; then + echo -e "${GREEN}✅ 集群连接正常${NC}" +else + echo -e "${RED}❌ 无法连接到集群${NC}" + exit 1 +fi + +# 检查命名空间 +echo "" +echo "2. 检查命名空间..." +if kubectl get namespace swarm-system &> /dev/null; then + echo -e "${GREEN}✅ swarm-system 命名空间存在${NC}" +else + echo -e "${RED}❌ swarm-system 命名空间不存在${NC}" + exit 1 +fi + +# 检查 Redis +echo "" +echo "3. 检查 Redis..." +REDIS_STATUS=$(kubectl get pod redis-0 -n swarm-system -o jsonpath='{.status.phase}' 2>/dev/null) +if [ "$REDIS_STATUS" = "Running" ]; then + echo -e "${GREEN}✅ Redis 运行中${NC}" +else + echo -e "${RED}❌ Redis 状态: $REDIS_STATUS${NC}" +fi + +# 检查 Orchestrator +echo "" +echo "4. 检查 Orchestrator..." +ORCH_READY=$(kubectl get deployment orchestrator -n swarm-system -o jsonpath='{.status.readyReplicas}' 2>/dev/null) +if [ "$ORCH_READY" = "1" ]; then + echo -e "${GREEN}✅ Orchestrator 运行中${NC}" +else + echo -e "${RED}❌ Orchestrator 未就绪${NC}" +fi + +# 获取 Orchestrator 外部 IP +echo "" +echo "5. 获取访问地址..." +EXTERNAL_IP=$(kubectl get svc orchestrator-service -n swarm-system -o jsonpath='{.status.loadBalancer.ingress[0].ip}' 2>/dev/null) +if [ -n "$EXTERNAL_IP" ]; then + echo -e "${GREEN}✅ Orchestrator 地址: http://$EXTERNAL_IP:8000${NC}" + + # 测试健康检查 + echo "" + echo "6. 测试 API 健康检查..." + if curl -s -f "http://$EXTERNAL_IP:8000/health" > /dev/null 2>&1; then + echo -e "${GREEN}✅ API 健康检查通过${NC}" + + # 显示健康状态详情 + HEALTH=$(curl -s "http://$EXTERNAL_IP:8000/health") + echo " 详情: $HEALTH" + else + echo -e "${YELLOW}⚠️ API 健康检查失败（可能还在启动中）${NC}" + fi +else + echo -e "${YELLOW}⚠️ 等待 LoadBalancer 分配外部 IP...${NC}" +fi + +# 显示所有 Pod +echo "" +echo "7. 所有 Pod 状态:" +echo "-----------------------------------" +kubectl get pods -n swarm-system + +# 显示服务 +echo "" +echo "8. 服务状态:" +echo "-----------------------------------" +kubectl get svc -n swarm-system + +# 总结 +echo "" +echo "======================================" +echo " 状态检查完成" +echo "======================================" +echo "" +echo "快速命令:" +echo " 查看日志: kubectl logs -n swarm-system deployment/orchestrator -f" +echo " 查看 Pod: kubectl get pods -n swarm-system" +echo " 运行测试: python3 test_system.py" +echo "" +echo "文档:" +echo " 快速入门: 快速入门指南.md" +echo " 完整文档: 部署和使用文档.md" +echo "" diff --git a/desktop-client/README.md b/desktop-client/README.md new file mode 100644 index 0000000..f38da9a --- /dev/null +++ b/desktop-client/README.md @@ -0,0 +1,71 @@ +# HeiCode Swarm 桌面客户端 + +基于 **Electron + React + TypeScript** 的桌面客户端，是蜂群系统面向用户的入口。它负责提交任务、连接并展示 Orchestrator 的运行状态，分析任务复杂度以推荐 Agent 数量，并对生成结果进行聚合与查看。 + +## 环境要求 + +- Node.js 18+ 与 npm +- 可访问的 Kubernetes 集群与有效的 kubeconfig（如使用 K8s 调度） +- 已正确配置的 kubectl + +## 安装与开发 + +```bash +npm install # 安装依赖 +npm run build # 编译 TypeScript +npm run dev # 开发模式启动 +npm run type-check +npm run lint +``` + +## 配置 + +应用按以下顺序加载 kubeconfig： +1. 环境变量 `KUBECONFIG`（若设置） +2. 默认位置 `~/.kube/config` + +请确保 kubeconfig 指向正确的集群且凭据有效。 + +## 目录结构 + +``` +desktop-client/ +├── src/ +│ ├── main.ts # Electron 主进程 +│ ├── App.tsx # React 根组件 +│ ├── index.html # 入口页面 +│ ├── api/ +│ │ └── orchestrator-client.ts # 调用 Orchestrator REST 接口 +│ ├── ai/ # 任务复杂度分析（见该目录 README） +│ ├── components/ # 界面组件 +│ │ ├── TaskSubmissionForm.tsx # 任务提交 +│ │ ├── AgentDashboard.tsx # Agent 与运行状态看板 +│ │ ├── ResultViewer.tsx # 结果查看 +│ │ └── CostControlDialog.tsx # 成本/规模确认 +│ ├── analytics/usage-tracker.ts # 用量统计 +│ ├── git/ # Git 工作区与分支合并 +│ │ ├── workspace-manager.ts +│ │ └── merge-manager.ts +│ └── k8s/ # K8s 接入 +│ ├── kubeconfig-loader.ts # 鉴权/集群配置加载 +│ └── auto-scaler.ts # 按复杂度自动伸缩 Agent 规模 +├── package.json +├── tsconfig.json +└── tailwind.config.js +``` + +## 主要能力 + +- **任务提交**：用户输入需求后，通过 `orchestrator-client` 创建蜂群部署。 +- **复杂度分析**：`ai/` 模块评估任务规模并推荐 Agent 数量（详见 `src/ai/README.md`）。 +- **成本控制**：超过阈值时通过 `CostControlDialog` 提示并要求确认，避免规模失控。 +- **状态展示**：`AgentDashboard` 实时展示 Agent、任务与工作流阶段。 +- **结果聚合**：`git/merge-manager` 合并各 Agent 的结果分支，`ResultViewer` 统一查看。 +- **弹性伸缩**：`k8s/auto-scaler` 依据复杂度分析结果调整 Agent 数量。 + +## 工作方式 + +1. 用户在客户端提交需求并（可选）查看复杂度分析与推荐规模。 +2. 客户端调用 Orchestrator 创建部署，Orchestrator 分解任务并调度 Agent 执行。 +3. 客户端轮询/订阅运行状态，展示阶段进度与各 Agent 产出。 +4. 运行完成后，聚合各结果分支并在结果视图中呈现最终交付物。 diff --git a/desktop-client/package.json b/desktop-client/package.json new file mode 100644 index 0000000..5011344 --- /dev/null +++ b/desktop-client/package.json @@ -0,0 +1,72 @@ +{ + "name": "heicode-swarm-desktop", + "version": "0.1.0", + "description": "Desktop client for HeiCode Swarm - K8s-based AI agent orchestration", + "main": "dist/main.js", + "scripts": { + "start": "electron .", + "dev": "concurrently \"npm run build:watch\" \"npm run electron:dev\"", + "build": "tsc", + "build:watch": "tsc --watch", + "electron:dev": "wait-on dist/main.js && electron .", + "package": "electron-builder", + "lint": "eslint src --ext .ts,.tsx", + "type-check": "tsc --noEmit", + "validate-analyzer": "tsc && node dist/ai/validate-analyzer.js" + }, + "keywords": [ + "electron", + "kubernetes", + "ai", + "swarm", + "orchestration" + ], + "author": "HeiCode", + "license": "MIT", + "dependencies": { + "@kubernetes/client-node": "^0.20.0", + "@anthropic-ai/sdk": "^0.20.0", + "isomorphic-git": "^1.25.0", + "zustand": "^4.5.0", + "react": "^18.2.0", + "react-dom": "^18.2.0" + }, + "devDependencies": { + "@types/node": "^20.11.0", + "@types/react": "^18.2.0", + "@types/react-dom": "^18.2.0", + "@typescript-eslint/eslint-plugin": "^6.19.0", + "@typescript-eslint/parser": "^6.19.0", + "concurrently": "^8.2.2", + "electron": "^28.0.0", + "electron-builder": "^24.9.1", + "eslint": "^8.56.0", + "typescript": "^5.3.3", + "wait-on": "^7.2.0", + "tailwindcss": "^3.4.0", + "autoprefixer": "^10.4.17", + "postcss": "^8.4.33" + }, + "build": { + "appId": "com.heicode.swarm", + "productName": "HeiCode Swarm", + "directories": { + "output": "release" + }, + "files": [ + "dist/**/*", + "package.json" + ], + "mac": { + "category": "public.app-category.developer-tools", + "target": ["dmg", "zip"] + }, + "linux": { + "target": ["AppImage", "deb"], + "category": "Development" + }, + "win": { + "target": ["nsis", "portable"] + } + } +} diff --git a/desktop-client/src/App.tsx b/desktop-client/src/App.tsx new file mode 100644 index 0000000..9d26b5c --- /dev/null +++ b/desktop-client/src/App.tsx @@ -0,0 +1,319 @@ +import React, { useState, useEffect } from 'react'; +import { createRoot } from 'react-dom/client'; +import { TaskSubmissionForm } from './components/TaskSubmissionForm'; +import { AgentDashboard } from './components/AgentDashboard'; +import { ResultViewer } from './components/ResultViewer'; +import { ComplexityAnalysis, getAnalyzer } from './ai/complexity-analyzer'; +import { getOrchestratorClient } from './api/orchestrator-client'; +import { AutoScaler } from './k8s/auto-scaler'; +import { MergeManager, MergeConflict, ConflictResolution } from './git/merge-manager'; +import { WorkspaceManager } from './git/workspace-manager'; +import { KubeConfigLoader } from './k8s/kubeconfig-loader'; + +interface AppState { + status: 'idle' | 'loading' | 'connected' | 'error'; + message: string; + currentWorkspaceId: string | null; + mergeConflicts: MergeConflict[]; + mergedBranches: string[]; + showConflictResolution: boolean; +} + +const App: React.FC = () => { + const [state, setState] = useState({ + status: 'idle', + message: 'Initializing HeiCode Swarm...', + currentWorkspaceId: null, + mergeConflicts: [], + mergedBranches: [], + showConflictResolution: false, + }); + + const [orchestratorUrl] = useState('http://localhost:8000'); + const [autoScaler, setAutoScaler] = useState(null); + const [workspaceManager] = useState(new WorkspaceManager()); + + useEffect(() => { + initializeApp(); + }, []); + + const initializeApp = async () => { + setState(prev => ({ ...prev, status: 'loading', message: 'Connecting to Kubernetes...' })); + + try { + // Load kubeconfig + const kubeLoader = new KubeConfigLoader(); + kubeLoader.load(); + + const validation = kubeLoader.validate(); + if (!validation.valid) { + throw new Error(`Kubeconfig validation failed: ${validation.error}`); + } + + // Test connection + const connectionTest = await kubeLoader.testConnection(); + if (!connectionTest.success) { + throw new Error(`K8s connection failed: ${connectionTest.error}`); + } + + // Initialize auto-scaler + const scaler = new AutoScaler(kubeLoader.getKubeConfig(), 'default'); + setAutoScaler(scaler); + + // Initialize complexity analyzer + const apiKey = process.env.ANTHROPIC_API_KEY || ''; + if (!apiKey) { + console.warn('ANTHROPIC_API_KEY not set. Complexity analysis will fail.'); + } + getAnalyzer({ apiKey }); + + // Initialize orchestrator client + getOrchestratorClient({ baseUrl: orchestratorUrl }); + + setState(prev => ({ + ...prev, + status: 'connected', + message: `Connected to ${validation.cluster}`, + })); + } catch (error) { + setState(prev => ({ + ...prev, + status: 'error', + message: error instanceof Error ? error.message : 'Initialization failed', + })); + } + }; + + const handleTaskSubmit = async (taskDescription: string, complexity: ComplexityAnalysis) => { + if (!autoScaler) { + alert('Auto-scaler not initialized'); + return; + } + + setState(prev => ({ ...prev, status: 'loading', message: 'Creating workspace...' })); + + try { + // Create workspace + const workspaceId = `ws-${Date.now()}`; + const workspaceResult = await workspaceManager.initWorkspace(workspaceId); + + if (!workspaceResult.success) { + throw new Error(`Failed to create workspace: ${workspaceResult.error}`); + } + + const { gitUrl } = workspaceResult.data; + + setState(prev => ({ ...prev, message: 'Creating agent pods...' })); + + // Scale up agents + const scaleResult = await autoScaler.scaleUp(complexity, { + namespace: 'default', + image: 'heicode-swarm-agent:latest', + orchestratorUrl, + gitUrl, + workspaceId, + }); + + if (!scaleResult.success) { + throw new Error(`Failed to create agents: ${scaleResult.errors.join(', ')}`); + } + + setState(prev => ({ ...prev, message: 'Submitting task to orchestrator...' })); + + // Create task in orchestrator + const client = getOrchestratorClient(); + await client.createTask(taskDescription, { + workspaceId, + complexity: complexity.complexity, + agentCount: complexity.agentCount, + }); + + setState(prev => ({ + ...prev, + status: 'connected', + message: `Task submitted with ${scaleResult.createdPods.length} agents`, + currentWorkspaceId: workspaceId, + })); + } catch (error) { + setState(prev => ({ + ...prev, + status: 'error', + message: error instanceof Error ? error.message : 'Task submission failed', + })); + } + }; + + const handleMergeResults = async () => { + if (!state.currentWorkspaceId) { + alert('No active workspace'); + return; + } + + setState(prev => ({ ...prev, status: 'loading', message: 'Merging agent results...' })); + + try { + const repoPath = workspaceManager.getWorkspacePath(state.currentWorkspaceId); + const mergeManager = new MergeManager(repoPath); + + // List agent branches + const agentBranches = await mergeManager.listAgentBranches(state.currentWorkspaceId); + + if (agentBranches.length === 0) { + throw new Error('No agent branches found'); + } + + // Merge branches + const mergeResult = await mergeManager.mergeAgentResults('main', agentBranches); + + if (mergeResult.conflicts.length > 0) { + // Show conflict resolution UI + setState(prev => ({ + ...prev, + status: 'connected', + message: `${mergeResult.conflicts.length} conflicts detected`, + mergeConflicts: mergeResult.conflicts, + mergedBranches: mergeResult.mergedBranches, + showConflictResolution: true, + })); + } else { + // Merge successful + setState(prev => ({ + ...prev, + status: 'connected', + message: 'Merge completed successfully', + mergedBranches: mergeResult.mergedBranches, + showConflictResolution: false, + })); + } + } catch (error) { + setState(prev => ({ + ...prev, + status: 'error', + message: error instanceof Error ? error.message : 'Merge failed', + })); + } + }; + + const handleConflictResolve = async (resolutions: ConflictResolution[]) => { + if (!state.currentWorkspaceId) return; + + setState(prev => ({ ...prev, status: 'loading', message: 'Applying resolutions...' })); + + try { + const repoPath = workspaceManager.getWorkspacePath(state.currentWorkspaceId); + const mergeManager = new MergeManager(repoPath); + + const success = await mergeManager.resolveConflicts(resolutions); + + if (success) { + setState(prev => ({ + ...prev, + status: 'connected', + message: 'Conflicts resolved successfully', + mergeConflicts: [], + showConflictResolution: false, + })); + } else { + throw new Error('Failed to apply resolutions'); + } + } catch (error) { + setState(prev => ({ + ...prev, + status: 'error', + message: error instanceof Error ? error.message : 'Resolution failed', + })); + } + }; + + const handleConflictCancel = () => { + setState(prev => ({ + ...prev, + mergeConflicts: [], + showConflictResolution: false, + })); + }; + + return ( +
+
+

HeiCode Swarm

+

K8s-Based AI Agent Orchestration

+
+ +
+
+
+
+
+

+ {state.status === 'connected' ? 'Connected' : + state.status === 'error' ? 'Error' : + state.status === 'loading' ? 'Loading...' : + 'Idle'} +

+

{state.message}

+
+
+
+ + {state.showConflictResolution ? ( + + ) : ( +
+
+ + + {state.currentWorkspaceId && ( +
+

Workspace Actions

+ +
+ )} +
+ +
+ +
+
+ )} +
+ +
+

+ Phase 5: Desktop Client & Orchestration - Full Integration +

+
+
+ ); +}; + +// Initialize React app +const container = document.getElementById('root'); +if (container) { + const root = createRoot(container); + root.render(); +} + +export default App; diff --git a/desktop-client/src/ai/README.md b/desktop-client/src/ai/README.md new file mode 100644 index 0000000..b01a6c2 --- /dev/null +++ b/desktop-client/src/ai/README.md @@ -0,0 +1,79 @@ +# AI 模块 —— 任务复杂度分析 + +该模块为桌面客户端提供**任务复杂度分析**能力：根据用户需求评估其规模与难度，推荐应投入的 Agent 数量，作为蜂群规模与成本控制的依据。 + +## 组成 + +### 1. `complexity-analyzer.ts` +核心分析器，调用大模型对任务进行复杂度评估并给出推荐 Agent 数量。 + +特性： +- 结构化 JSON 结果解析，并对非 JSON 响应做兜底解析 +- Agent 数量校验（1–50） +- 性能监控（目标单次 < 3 秒） +- 支持批量分析 +- 单例模式复用客户端 + +用法： +```typescript +import { ComplexityAnalyzer } from './ai'; + +const analyzer = new ComplexityAnalyzer({ temperature: 0.3 }); +const analysis = await analyzer.analyzeTask('为整个项目补充单元测试'); +console.log(`推荐 Agent 数：${analysis.agentCount}`); +console.log(`复杂度：${analysis.complexity}`); +console.log(`理由：${analysis.reasoning}`); +``` + +### 2. `prompt-builder.ts` +基于带标注的数据集构造 few-shot 提示词。 + +特性： +- 从 `test-data/complexity-dataset.json` 加载样例 +- 按比例选取多样化样例（低 40% / 中 40% / 高 20%） +- 生成完整的 few-shot 提示，批量模式下提供简化提示 +- 数据集不可用时回退到内置默认样例 + +### 3. `validator.ts` +对照标注数据集校验分析器的准确率。 + +特性与验收目标： +- 全量与快速（10 条抽样）校验 +- 准确率（精确匹配、±2 个 Agent 容差）与平均绝对误差（MAE） +- 不匹配项报告与 Markdown 报告生成 +- 目标：准确率 ≥ 70%（±2 容差）、MAE ≤ 2、单任务分析 < 3 秒 + +### 4. `validate-analyzer.ts` +独立的校验脚本，便于在集成前验证分析器质量： +```bash +npm run validate-analyzer +``` + +## 成本控制 + +分析结果与 `CostControlDialog` 组件联动，向用户提示资源开销： +- **> 10 个 Agent**：提示并给出资源预估 +- **> 15 个 Agent**：要求用户显式确认 +- **达到上限**：遵循用户配置的最大值（默认 20） + +## 数据集格式 + +标注数据集 `test-data/complexity-dataset.json` 的样例格式： + +```json +[ + { + "task": "修复 main.py 第 42 行的语法错误", + "expectedAgents": 1, + "complexity": "low", + "reasoning": "单文件单行修复，无依赖与副作用。" + } +] +``` + +## 实现要点 + +1. **Few-shot 学习**：使用数据集中的多样化样例提升判断稳定性。 +2. **温度 0.3**：降低随机性以获得更一致的预测。 +3. **稳健性**：解析失败时优雅兜底。 +4. **先验证后上线**：集成进自动伸缩与任务提交流程前，应先通过校验。 diff --git a/desktop-client/src/ai/complexity-analyzer.test.ts b/desktop-client/src/ai/complexity-analyzer.test.ts new file mode 100644 index 0000000..c632c80 --- /dev/null +++ b/desktop-client/src/ai/complexity-analyzer.test.ts @@ -0,0 +1,120 @@ +/** + * Unit tests for complexity analyzer + * Run with: npm test + */ + +import { ComplexityAnalyzer } from './complexity-analyzer'; +import { loadExamples, selectFewShotExamples, buildComplexityPrompt } from './prompt-builder'; + +describe('ComplexityAnalyzer', () => { + let analyzer: ComplexityAnalyzer; + + beforeEach(() => { + analyzer = new ComplexityAnalyzer({ + apiKey: process.env.ANTHROPIC_API_KEY || 'test-key', + model: 'claude-opus-4-7', + }); + }); + + describe('validateAgentCount', () => { + it('should clamp negative values to 1', () => { + const result = (analyzer as any).validateAgentCount(-5); + expect(result).toBe(1); + }); + + it('should clamp values above 50', () => { + const result = (analyzer as any).validateAgentCount(100); + expect(result).toBe(50); + }); + + it('should round decimal values', () => { + const result = (analyzer as any).validateAgentCount(3.7); + expect(result).toBe(4); + }); + + it('should accept valid values', () => { + const result = (analyzer as any).validateAgentCount(10); + expect(result).toBe(10); + }); + }); + + describe('validateComplexity', () => { + it('should accept valid complexity levels', () => { + expect((analyzer as any).validateComplexity('low')).toBe('low'); + expect((analyzer as any).validateComplexity('medium')).toBe('medium'); + expect((analyzer as any).validateComplexity('high')).toBe('high'); + }); + + it('should default to medium for invalid values', () => { + expect((analyzer as any).validateComplexity('invalid')).toBe('medium'); + }); + }); + + describe('inferComplexity', () => { + it('should infer low complexity for 1-2 agents', () => { + expect((analyzer as any).inferComplexity(1)).toBe('low'); + expect((analyzer as any).inferComplexity(2)).toBe('low'); + }); + + it('should infer medium complexity for 3-7 agents', () => { + expect((analyzer as any).inferComplexity(3)).toBe('medium'); + expect((analyzer as any).inferComplexity(7)).toBe('medium'); + }); + + it('should infer high complexity for 8+ agents', () => { + expect((analyzer as any).inferComplexity(8)).toBe('high'); + expect((analyzer as any).inferComplexity(20)).toBe('high'); + }); + }); +}); + +describe('PromptBuilder', () => { + describe('loadExamples', () => { + it('should load examples from dataset', () => { + const examples = loadExamples(); + expect(examples.length).toBeGreaterThan(0); + expect(examples[0]).toHaveProperty('task'); + expect(examples[0]).toHaveProperty('expectedAgents'); + expect(examples[0]).toHaveProperty('complexity'); + expect(examples[0]).toHaveProperty('reasoning'); + }); + }); + + describe('selectFewShotExamples', () => { + it('should select requested number of examples', () => { + const examples = loadExamples(); + const selected = selectFewShotExamples(examples, 12); + expect(selected.length).toBeLessThanOrEqual(12); + }); + + it('should include diverse complexity levels', () => { + const examples = loadExamples(); + const selected = selectFewShotExamples(examples, 12); + const complexities = selected.map((e) => e.complexity); + expect(complexities).toContain('low'); + expect(complexities).toContain('medium'); + expect(complexities).toContain('high'); + }); + }); + + describe('buildComplexityPrompt', () => { + it('should build prompt with task description', () => { + const prompt = buildComplexityPrompt('Fix syntax error'); + expect(prompt).toContain('Fix syntax error'); + }); + + it('should include few-shot examples', () => { + const prompt = buildComplexityPrompt('Test task'); + expect(prompt).toContain('Task:'); + expect(prompt).toContain('Analysis:'); + expect(prompt).toContain('agentCount'); + }); + + it('should include guidelines', () => { + const prompt = buildComplexityPrompt('Test task'); + expect(prompt).toContain('Low complexity'); + expect(prompt).toContain('Medium complexity'); + expect(prompt).toContain('High complexity'); + }); + }); +}); diff --git a/desktop-client/src/ai/complexity-analyzer.ts b/desktop-client/src/ai/complexity-analyzer.ts new file mode 100644 index 0000000..7889890 --- /dev/null +++ b/desktop-client/src/ai/complexity-analyzer.ts @@ -0,0 +1,191 @@ +import Anthropic from '@anthropic-ai/sdk'; +import { buildComplexityPrompt } from './prompt-builder'; + +export interface ComplexityAnalysis { + agentCount: number; + reasoning: string; + complexity: 'low' | 'medium' | 'high'; + confidence: number; +} + +export interface AnalyzerConfig { + apiKey: string; + model?: string; + maxTokens?: number; + temperature?: number; +} + +export class ComplexityAnalyzer { + private client: Anthropic; + private model: string; + private maxTokens: number; + private temperature: number; + + constructor(config: AnalyzerConfig) { + this.client = new Anthropic({ + apiKey: config.apiKey, + }); + this.model = config.model || 'claude-opus-4-7'; + this.maxTokens = config.maxTokens || 1024; + this.temperature = config.temperature || 0.3; + } + + /** + * Analyze task complexity and determine required agent count + * @param taskDescription - The task to analyze + * @returns ComplexityAnalysis with agent count recommendation + */ + async analyzeTask(taskDescription: string): Promise { + const startTime = Date.now(); + + try { + const prompt = buildComplexityPrompt(taskDescription); + + const response = await this.client.messages.create({ + model: this.model, + max_tokens: this.maxTokens, + temperature: this.temperature, + messages: [ + { + role: 'user', + content: prompt, + }, + ], + }); + + const elapsedTime = Date.now() - startTime; + console.log(`Complexity analysis completed in ${elapsedTime}ms`); + + if (elapsedTime > 3000) { + console.warn(`Analysis took ${elapsedTime}ms, exceeding 3s target`); + } + + const content = response.content[0]; + if (content.type !== 'text') { + throw new Error('Unexpected response type from Claude API'); + } + + const analysis = this.parseResponse(content.text); + return analysis; + } catch (error) { + console.error('Error analyzing task complexity:', error); + throw new Error(`Failed to analyze task complexity: ${error instanceof Error ? error.message : 'Unknown error'}`); + } + } + + /** + * Parse Claude's response into structured ComplexityAnalysis + */ + private parseResponse(responseText: string): ComplexityAnalysis { + try { + // Try to extract JSON from response + const jsonMatch = responseText.match(/\{[\s\S]*\}/); + if (jsonMatch) { + const parsed = JSON.parse(jsonMatch[0]); + return { + agentCount: this.validateAgentCount(parsed.agentCount), + reasoning: parsed.reasoning || 'No reasoning provided', + complexity: this.validateComplexity(parsed.complexity), + confidence: parsed.confidence || 0.8, + }; + } + + // Fallback: try to extract agent count from text + const agentMatch = responseText.match(/(\d+)\s*agent/i); + if (agentMatch) { + const agentCount = parseInt(agentMatch[1], 10); + return { + agentCount: this.validateAgentCount(agentCount), + reasoning: responseText, + complexity: this.inferComplexity(agentCount), + confidence: 0.6, + }; + } + + throw new Error('Could not parse agent count from response'); + } catch (error) { + console.error('Error parsing response:', responseText); + throw new Error(`Failed to parse complexity analysis: ${error instanceof Error ? error.message : 'Unknown error'}`); + } + } + + /** + * Validate and clamp agent count to reasonable range + */ + private validateAgentCount(count: number): number { + if (isNaN(count) || count < 1) { + console.warn(`Invalid agent count ${count}, defaulting to 1`); + return 1; + } + if (count > 50) { + console.warn(`Agent count ${count} exceeds maximum, capping at 50`); + return 50; + } + return Math.round(count); + } + + /** + * Validate complexity level + */ + private validateComplexity(complexity: string): 'low' | 'medium' | 'high' { + const normalized = complexity.toLowerCase(); + if (normalized === 'low' || normalized === 'medium' || normalized === 'high') { + return normalized as 'low' | 'medium' | 'high'; + } + console.warn(`Invalid complexity level ${complexity}, defaulting to medium`); + return 'medium'; + } + + /** + * Infer complexity from agent count + */ + private inferComplexity(agentCount: number): 'low' | 'medium' | 'high' { + if (agentCount <= 2) return 'low'; + if (agentCount <= 7) return 'medium'; + return 'high'; + } + + /** + * Batch analyze multiple tasks + */ + async analyzeBatch(tasks: string[]): Promise { + const results: ComplexityAnalysis[] = []; + + for (const task of tasks) { + try { + const analysis = await this.analyzeTask(task); + results.push(analysis); + } catch (error) { + console.error(`Failed to analyze task: ${task}`, error); + // Push a default analysis for failed tasks + results.push({ + agentCount: 1, + reasoning: `Analysis failed: ${error instanceof Error ? error.message : 'Unknown error'}`, + complexity: 'low', + confidence: 0, + }); + } + } + + return results; + } +} + +/** + * Create a singleton analyzer instance + */ +let analyzerInstance: ComplexityAnalyzer | null = null; + +export function getAnalyzer(config?: AnalyzerConfig): ComplexityAnalyzer { + if (!analyzerInstance) { + if (!config) { + throw new Error('Analyzer not initialized. Provide config on first call.'); + } + analyzerInstance = new ComplexityAnalyzer(config); + } + return analyzerInstance; +} + +export function resetAnalyzer(): void { + analyzerInstance = null; +} diff --git a/desktop-client/src/ai/example-usage.ts b/desktop-client/src/ai/example-usage.ts new file mode 100644 index 0000000..31b6448 --- /dev/null +++ b/desktop-client/src/ai/example-usage.ts @@ -0,0 +1,77 @@ +/** + * Example usage of the complexity analyzer + * This demonstrates how to use the analyzer in the desktop client + */ + +import { ComplexityAnalyzer, validateAnalyzer } from './index'; + +async function exampleUsage() { + // Initialize analyzer + const analyzer = new ComplexityAnalyzer({ + apiKey: process.env.ANTHROPIC_API_KEY || 'your-api-key', + model: 'claude-opus-4-7', + temperature: 0.3, + }); + + // Example 1: Analyze a simple task + console.log('=== Example 1: Simple Task ==='); + const simpleTask = 'Fix syntax error in main.py line 42'; + const simpleAnalysis = await analyzer.analyzeTask(simpleTask); + console.log(`Task: ${simpleTask}`); + console.log(`Agents needed: ${simpleAnalysis.agentCount}`); + console.log(`Complexity: ${simpleAnalysis.complexity}`); + console.log(`Reasoning: ${simpleAnalysis.reasoning}\n`); + + // Example 2: Analyze a medium complexity task + console.log('=== Example 2: Medium Complexity Task ==='); + const mediumTask = 'Add input validation to all API endpoints'; + const mediumAnalysis = await analyzer.analyzeTask(mediumTask); + console.log(`Task: ${mediumTask}`); + console.log(`Agents needed: ${mediumAnalysis.agentCount}`); + console.log(`Complexity: ${mediumAnalysis.complexity}`); + console.log(`Reasoning: ${mediumAnalysis.reasoning}\n`); + + // Example 3: Analyze a complex task + console.log('=== Example 3: Complex Task ==='); + const complexTask = 'Add comprehensive unit tests to the entire project'; + const complexAnalysis = await analyzer.analyzeTask(complexTask); + console.log(`Task: ${complexTask}`); + console.log(`Agents needed: ${complexAnalysis.agentCount}`); + console.log(`Complexity: ${complexAnalysis.complexity}`); + console.log(`Reasoning: ${complexAnalysis.reasoning}\n`); + + // Example 4: Batch analysis + console.log('=== Example 4: Batch Analysis ==='); + const tasks = [ + 'Update README.md with installation instructions', + 'Implement caching layer using Redis', + 'Migrate from REST API to GraphQL', + ]; + const batchResults = await analyzer.analyzeBatch(tasks); + batchResults.forEach((result, i: number) => { + console.log(`${i + 1}. ${tasks[i]}: ${result.agentCount} agents (${result.complexity})`); + }); + console.log(); + + // Example 5: Run validation (optional) + if (process.env.RUN_VALIDATION === 'true') { + console.log('=== Example 5: Validation ==='); + const validationResult = await validateAnalyzer(analyzer, { + accuracyThreshold: 0.7, + maxMeanAbsoluteError: 2, + }); + console.log(`Validation ${validationResult.passed ? 'PASSED' : 'FAILED'}`); + console.log(`Accuracy: ${(validationResult.accuracy * 100).toFixed(1)}%`); + console.log(`MAE: ${validationResult.meanAbsoluteError.toFixed(2)}`); + } +} + +// Run if executed directly +if (require.main === module) { + exampleUsage().catch((error) => { + console.error('Error:', error); + process.exit(1); + }); +} + +export { exampleUsage }; diff --git a/desktop-client/src/ai/index.ts b/desktop-client/src/ai/index.ts new file mode 100644 index 0000000..fd9948b --- /dev/null +++ b/desktop-client/src/ai/index.ts @@ -0,0 +1,10 @@ +/** + * AI Module - Task Complexity Analysis + * + * This module provides AI-powered task complexity analysis using Claude Opus 4.7. + * It determines the optimal number of agents needed to complete a given task. + */ + +export { ComplexityAnalyzer, ComplexityAnalysis, AnalyzerConfig, getAnalyzer, resetAnalyzer } from './complexity-analyzer'; +export { loadExamples, selectFewShotExamples, buildComplexityPrompt, buildSimplifiedPrompt, ComplexityExample } from './prompt-builder'; +export { validateAnalyzer, quickValidation, generateValidationReport, ValidationResult, ValidationMismatch, ValidationConfig } from './validator'; diff --git a/desktop-client/src/ai/prompt-builder.ts b/desktop-client/src/ai/prompt-builder.ts new file mode 100644 index 0000000..0e98416 --- /dev/null +++ b/desktop-client/src/ai/prompt-builder.ts @@ -0,0 +1,172 @@ +import * as fs from 'fs'; +import * as path from 'path'; + +export interface ComplexityExample { + task: string; + expectedAgents: number; + complexity: 'low' | 'medium' | 'high'; + reasoning: string; +} + +/** + * Load complexity examples from the test dataset + */ +export function loadExamples(datasetPath?: string): ComplexityExample[] { + const defaultPath = path.join(__dirname, '../../../test-data/complexity-dataset.json'); + const filePath = datasetPath || defaultPath; + + try { + const fileContent = fs.readFileSync(filePath, 'utf-8'); + const examples = JSON.parse(fileContent) as ComplexityExample[]; + return examples; + } catch (error) { + console.error(`Failed to load examples from ${filePath}:`, error); + return getDefaultExamples(); + } +} + +/** + * Get default examples if dataset file is not available + */ +function getDefaultExamples(): ComplexityExample[] { + return [ + { + task: 'Fix syntax error in main.py line 42', + expectedAgents: 1, + complexity: 'low', + reasoning: 'Single file, single line fix. No dependencies or side effects.', + }, + { + task: 'Add type hints to the calculate_total function', + expectedAgents: 1, + complexity: 'low', + reasoning: 'Single function modification in one file.', + }, + { + task: 'Refactor the user authentication module to use JWT tokens', + expectedAgents: 3, + complexity: 'medium', + reasoning: 'Affects multiple functions in auth module, requires updating login/logout flows and adding token validation.', + }, + { + task: 'Add input validation to all API endpoints', + expectedAgents: 5, + complexity: 'medium', + reasoning: 'Multiple endpoints across different route files need validation logic.', + }, + { + task: 'Add comprehensive unit tests to the entire project', + expectedAgents: 12, + complexity: 'high', + reasoning: 'Testing all modules, functions, edge cases. Requires understanding entire codebase and writing hundreds of test cases.', + }, + { + task: 'Refactor entire codebase to follow clean architecture principles', + expectedAgents: 15, + complexity: 'high', + reasoning: 'Major restructuring: separate layers (domain, application, infrastructure), update all imports, maintain functionality.', + }, + ]; +} + +/** + * Select diverse examples for few-shot prompting + * Ensures representation across low, medium, and high complexity + */ +export function selectFewShotExamples( + examples: ComplexityExample[], + count: number = 12 +): ComplexityExample[] { + const lowExamples = examples.filter((e) => e.complexity === 'low'); + const mediumExamples = examples.filter((e) => e.complexity === 'medium'); + const highExamples = examples.filter((e) => e.complexity === 'high'); + + // Aim for balanced distribution: 40% low, 40% medium, 20% high + const lowCount = Math.ceil(count * 0.4); + const mediumCount = Math.ceil(count * 0.4); + const highCount = count - lowCount - mediumCount; + + const selected: ComplexityExample[] = []; + + // Randomly select from each category + selected.push(...selectRandom(lowExamples, lowCount)); + selected.push(...selectRandom(mediumExamples, mediumCount)); + selected.push(...selectRandom(highExamples, highCount)); + + return selected; +} + +/** + * Randomly select N items from array + */ +function selectRandom(array: T[], count: number): T[] { + const shuffled = [...array].sort(() => Math.random() - 0.5); + return shuffled.slice(0, Math.min(count, array.length)); +} + +/** + * Build the few-shot prompt for complexity analysis + */ +export function buildComplexityPrompt(taskDescription: string): string { + const examples = loadExamples(); + const fewShotExamples = selectFewShotExamples(examples, 12); + + const examplesText = fewShotExamples + .map( + (ex) => + `Task: "${ex.task}" +Analysis: { + "agentCount": ${ex.expectedAgents}, + "complexity": "${ex.complexity}", + "reasoning": "${ex.reasoning}", + "confidence": 0.9 +}` + ) + .join('\n\n'); + + return `You are an expert at analyzing software development task complexity and determining how many AI agents are needed to complete the task efficiently. + +Your job is to analyze a task description and return a JSON object with: +- agentCount: number of agents needed (1-50) +- complexity: "low", "medium", or "high" +- reasoning: brief explanation of your decision +- confidence: your confidence level (0.0-1.0) + +Guidelines for agent count estimation: +- **Low complexity (1-2 agents)**: Single file changes, simple fixes, documentation updates, small additions +- **Medium complexity (3-7 agents)**: Multi-file changes, refactoring modules, adding features across several files, implementing new subsystems +- **High complexity (8-20 agents)**: Major architectural changes, full test suites, large-scale refactoring, migrating technologies, building new systems + +Consider these factors: +1. **Scope**: How many files/modules are affected? +2. **Interdependencies**: Do changes require coordination across multiple components? +3. **Complexity**: Does it require deep understanding of the codebase? +4. **Testing**: How much testing is needed? +5. **Risk**: What's the blast radius of potential errors? + +Here are examples of task complexity analysis: + +${examplesText} + +Now analyze this task: + +Task: "${taskDescription}" + +Return ONLY a JSON object with the analysis. No additional text.`; +} + +/** + * Build a simplified prompt for faster analysis (used in batch mode) + */ +export function buildSimplifiedPrompt(taskDescription: string): string { + return `Analyze this software development task and estimate how many AI agents (1-50) are needed to complete it. + +Task: "${taskDescription}" + +Consider: +- Single file/function changes = 1 agent +- Multi-file changes or module refactoring = 3-7 agents +- Major architectural changes or full test suites = 8-20 agents + +Return JSON: {"agentCount": N, "complexity": "low|medium|high", "reasoning": "brief explanation"}`; +} diff --git a/desktop-client/src/ai/validate-analyzer.ts b/desktop-client/src/ai/validate-analyzer.ts new file mode 100644 index 0000000..5cd3e8c --- /dev/null +++ b/desktop-client/src/ai/validate-analyzer.ts @@ -0,0 +1,57 @@ +#!/usr/bin/env node +/** + * Validation script for complexity analyzer + * Run with: npm run validate-analyzer + */ + +import { ComplexityAnalyzer } from './complexity-analyzer'; +import { validateAnalyzer, generateValidationReport } from './validator'; +import * as fs from 'fs'; +import * as path from 'path'; + +async function main() { + console.log('=== Complexity Analyzer Validation ===\n'); + + // Check for API key + const apiKey = process.env.ANTHROPIC_API_KEY; + if (!apiKey) { + console.error('Error: ANTHROPIC_API_KEY environment variable not set'); + console.error('Please set it with: export ANTHROPIC_API_KEY=your_key_here'); + process.exit(1); + } + + // Initialize analyzer + console.log('Initializing analyzer with Claude Opus 4.7...'); + const analyzer = new ComplexityAnalyzer({ + apiKey, + model: 'claude-opus-4-7', + temperature: 0.3, + }); + + // Run validation + const result = await validateAnalyzer(analyzer, { + accuracyThreshold: 0.7, + maxMeanAbsoluteError: 2, + }); + + // Generate report + const report = generateValidationReport(result); + const reportPath = path.join(__dirname, '../../../validation-report.md'); + fs.writeFileSync(reportPath, report); + console.log(`\nValidation report saved to: ${reportPath}`); + + // Exit with appropriate code + if (result.passed) { + console.log('\n✓ Validation PASSED - Analyzer meets acceptance criteria'); + process.exit(0); + } else { + console.log('\n✗ Validation FAILED - Analyzer does not meet acceptance criteria'); + console.log('Please review mismatches and tune the prompt.'); + process.exit(1); + } +} + +main().catch((error) => { + console.error('Validation failed with error:', error); + process.exit(1); +}); diff --git a/desktop-client/src/ai/validator.ts b/desktop-client/src/ai/validator.ts new file mode 100644 index 0000000..1298461 --- /dev/null +++ b/desktop-client/src/ai/validator.ts @@ -0,0 +1,247 @@ +import { ComplexityAnalyzer, ComplexityAnalysis } from './complexity-analyzer'; +import { loadExamples, ComplexityExample } from './prompt-builder'; + +export interface ValidationResult { + accuracy: number; + meanAbsoluteError: number; + totalTests: number; + correctPredictions: number; + withinTwoAgents: number; + mismatches: ValidationMismatch[]; + passed: boolean; +} + +export interface ValidationMismatch { + task: string; + expected: number; + predicted: number; + error: number; + reasoning: string; +} + +export interface ValidationConfig { + accuracyThreshold?: number; // Default: 0.7 (70%) + maxMeanAbsoluteError?: number; // Default: 2 + datasetPath?: string; +} + +/** + * Validate the complexity analyzer against the labeled dataset + */ +export async function validateAnalyzer( + analyzer: ComplexityAnalyzer, + config: ValidationConfig = {} +): Promise { + const accuracyThreshold = config.accuracyThreshold || 0.7; + const maxMAE = config.maxMeanAbsoluteError || 2; + + console.log('Loading test dataset...'); + const examples = loadExamples(config.datasetPath); + console.log(`Loaded ${examples.length} examples`); + + console.log('Running validation...'); + const startTime = Date.now(); + + const predictions: ComplexityAnalysis[] = []; + const mismatches: ValidationMismatch[] = []; + let totalError = 0; + let correctPredictions = 0; + let withinTwoAgents = 0; + + // Analyze each example + for (let i = 0; i < examples.length; i++) { + const example = examples[i]; + console.log(`[${i + 1}/${examples.length}] Analyzing: ${example.task.substring(0, 50)}...`); + + try { + const prediction = await analyzer.analyzeTask(example.task); + predictions.push(prediction); + + const error = Math.abs(prediction.agentCount - example.expectedAgents); + totalError += error; + + // Exact match + if (prediction.agentCount === example.expectedAgents) { + correctPredictions++; + withinTwoAgents++; + } + // Within ±2 agents + else if (error <= 2) { + withinTwoAgents++; + } + + // Record mismatches + if (error > 2) { + mismatches.push({ + task: example.task, + expected: example.expectedAgents, + predicted: prediction.agentCount, + error, + reasoning: prediction.reasoning, + }); + } + } catch (error) { + console.error(`Failed to analyze task: ${example.task}`, error); + // Count as maximum error + totalError += 50; + mismatches.push({ + task: example.task, + expected: example.expectedAgents, + predicted: 0, + error: 50, + reasoning: `Analysis failed: ${error instanceof Error ? error.message : 'Unknown error'}`, + }); + } + } + + const elapsedTime = Date.now() - startTime; + console.log(`Validation completed in ${elapsedTime}ms`); + + // Calculate metrics + const totalTests = examples.length; + const accuracy = withinTwoAgents / totalTests; + const meanAbsoluteError = totalError / totalTests; + + const passed = accuracy >= accuracyThreshold && meanAbsoluteError <= maxMAE; + + const result: ValidationResult = { + accuracy, + meanAbsoluteError, + totalTests, + correctPredictions, + withinTwoAgents, + mismatches, + passed, + }; + + // Print summary + console.log('\n=== Validation Results ==='); + console.log(`Total tests: ${totalTests}`); + console.log(`Exact matches: ${correctPredictions} (${((correctPredictions / totalTests) * 100).toFixed(1)}%)`); + console.log(`Within ±2 agents: ${withinTwoAgents} (${(accuracy * 100).toFixed(1)}%)`); + console.log(`Mean Absolute Error: ${meanAbsoluteError.toFixed(2)}`); + console.log(`Target accuracy: ${(accuracyThreshold * 100).toFixed(0)}%`); + console.log(`Target MAE: ≤${maxMAE}`); + console.log(`Status: ${passed ? '✓ PASSED' : '✗ FAILED'}`); + + if (mismatches.length > 0) { + console.log(`\n=== Mismatches (error >2 agents) ===`); + mismatches.slice(0, 10).forEach((m, i) => { + console.log(`\n${i + 1}. Task: ${m.task}`); + console.log(` Expected: ${m.expected}, Predicted: ${m.predicted}, Error: ${m.error}`); + console.log(` Reasoning: ${m.reasoning.substring(0, 100)}...`); + }); + if (mismatches.length > 10) { + console.log(`\n... and ${mismatches.length - 10} more mismatches`); + } + } + + return result; +} + +/** + * Run quick validation on a subset of examples + */ +export async function quickValidation( + analyzer: ComplexityAnalyzer, + sampleSize: number = 10 +): Promise { + const examples = loadExamples(); + + // Select diverse sample + const lowExamples = examples.filter((e) => e.complexity === 'low').slice(0, 3); + const mediumExamples = examples.filter((e) => e.complexity === 'medium').slice(0, 4); + const highExamples = examples.filter((e) => e.complexity === 'high').slice(0, 3); + + const sample = [...lowExamples, ...mediumExamples, ...highExamples]; + + console.log(`Running quick validation on ${sample.length} examples...`); + + const predictions: ComplexityAnalysis[] = []; + const mismatches: ValidationMismatch[] = []; + let totalError = 0; + let correctPredictions = 0; + let withinTwoAgents = 0; + + for (const example of sample) { + try { + const prediction = await analyzer.analyzeTask(example.task); + predictions.push(prediction); + + const error = Math.abs(prediction.agentCount - example.expectedAgents); + totalError += error; + + if (prediction.agentCount === example.expectedAgents) { + correctPredictions++; + withinTwoAgents++; + } else if (error <= 2) { + withinTwoAgents++; + } + + if (error > 2) { + mismatches.push({ + task: example.task, + expected: example.expectedAgents, + predicted: prediction.agentCount, + error, + reasoning: prediction.reasoning, + }); + } + } catch (error) { + console.error(`Failed to analyze task: ${example.task}`, error); + totalError += 50; + } + } + + const totalTests = sample.length; + const accuracy = withinTwoAgents / totalTests; + const meanAbsoluteError = totalError / totalTests; + + return { + accuracy, + meanAbsoluteError, + totalTests, + correctPredictions, + withinTwoAgents, + mismatches, + passed: accuracy >= 0.7 && meanAbsoluteError <= 2, + }; +} + +/** + * Generate a validation report as markdown + */ +export function generateValidationReport(result: ValidationResult): string { + const report = `# Complexity Analyzer Validation Report + +## Summary +- **Status**: ${result.passed ? '✓ PASSED' : '✗ FAILED'} +- **Total Tests**: ${result.totalTests} +- **Accuracy (±2 agents)**: ${(result.accuracy * 100).toFixed(1)}% +- **Exact Matches**: ${result.correctPredictions} (${((result.correctPredictions / result.totalTests) * 100).toFixed(1)}%) +- **Mean Absolute Error**: ${result.meanAbsoluteError.toFixed(2)} + +## Acceptance Criteria +- ✓ Accuracy ≥70%: ${result.accuracy >= 0.7 ? 'PASS' : 'FAIL'} +- ✓ MAE ≤2: ${result.meanAbsoluteError <= 2 ? 'PASS' : 'FAIL'} + +## Mismatches (Error >2 agents) +${result.mismatches.length === 0 ? 'None' : ''} +${result.mismatches + .map( + (m, i) => ` +### ${i + 1}. ${m.task} +- **Expected**: ${m.expected} agents +- **Predicted**: ${m.predicted} agents +- **Error**: ${m.error} agents +- **Reasoning**: ${m.reasoning} +` + ) + .join('\n')} + +--- +Generated: ${new Date().toISOString()} +`; + + return report; +} diff --git a/desktop-client/src/analytics/usage-tracker.ts b/desktop-client/src/analytics/usage-tracker.ts new file mode 100644 index 0000000..b4a7792 --- /dev/null +++ b/desktop-client/src/analytics/usage-tracker.ts @@ -0,0 +1,342 @@ +/** + * Usage Tracker for Agent Analytics and Cost Control + * + * Tracks agent usage, compute time, and estimated costs. + * Provides budget alerts and usage analytics. + */ + +export interface AgentUsageEvent { + agentId: string; + taskId: string; + eventType: 'created' | 'started' | 'completed' | 'failed' | 'terminated'; + timestamp: number; + metadata?: Record; +} + +export interface AgentSession { + agentId: string; + taskId: string; + startTime: number; + endTime?: number; + durationSeconds?: number; + podName: string; + nodeType: string; + cpuCores: number; + memoryMB: number; + status: 'running' | 'completed' | 'failed' | 'terminated'; + costUSD?: number; +} + +export interface UsageMetrics { + totalAgents: number; + activeAgents: number; + completedTasks: number; + failedTasks: number; + totalComputeHours: number; + totalCostUSD: number; + averageTaskDuration: number; + peakConcurrentAgents: number; +} + +export interface CostConfig { + cpuCostPerCoreHour: number; // USD per CPU core per hour + memoryCostPerGBHour: number; // USD per GB memory per hour + budgetLimitUSD?: number; + alertThresholds: number[]; // Alert at these percentages of budget +} + +export interface BudgetAlert { + timestamp: number; + currentSpend: number; + budgetLimit: number; + percentageUsed: number; + message: string; + severity: 'info' | 'warning' | 'critical'; +} + +export class UsageTracker { + private sessions: Map = new Map(); + private events: AgentUsageEvent[] = []; + private alerts: BudgetAlert[] = []; + private costConfig: CostConfig; + private alertedThresholds: Set = new Set(); + + constructor(costConfig: CostConfig) { + this.costConfig = costConfig; + } + + /** + * Track an agent lifecycle event + */ + trackEvent(event: AgentUsageEvent): void { + this.events.push(event); + + switch (event.eventType) { + case 'created': + this.handleAgentCreated(event); + break; + case 'started': + this.handleAgentStarted(event); + break; + case 'completed': + case 'failed': + case 'terminated': + this.handleAgentEnded(event); + break; + } + + // Check budget after each event + this.checkBudgetAlerts(); + } + + private handleAgentCreated(event: AgentUsageEvent): void { + const session: AgentSession = { + agentId: event.agentId, + taskId: event.taskId, + startTime: event.timestamp, + podName: event.metadata?.podName || '', + nodeType: event.metadata?.nodeType || 'standard', + cpuCores: event.metadata?.cpuCores || 2, + memoryMB: event.metadata?.memoryMB || 4096, + status: 'running' + }; + + this.sessions.set(event.agentId, session); + } + + private handleAgentStarted(event: AgentUsageEvent): void { + const session = this.sessions.get(event.agentId); + if (session) { + session.startTime = event.timestamp; + session.status = 'running'; + } + } + + private handleAgentEnded(event: AgentUsageEvent): void { + const session = this.sessions.get(event.agentId); + if (!session) return; + + session.endTime = event.timestamp; + session.durationSeconds = (event.timestamp - session.startTime) / 1000; + session.status = event.eventType === 'completed' ? 'completed' : + event.eventType === 'failed' ? 'failed' : 'terminated'; + + // Calculate cost + session.costUSD = this.calculateSessionCost(session); + } + + /** + * Calculate cost for an agent session + */ + private calculateSessionCost(session: AgentSession): number { + if (!session.durationSeconds) return 0; + + const hours = session.durationSeconds / 3600; + const cpuCost = session.cpuCores * this.costConfig.cpuCostPerCoreHour * hours; + const memoryGB = session.memoryMB / 1024; + const memoryCost = memoryGB * this.costConfig.memoryCostPerGBHour * hours; + + return cpuCost + memoryCost; + } + + /** + * Get current usage metrics + */ + getMetrics(): UsageMetrics { + const allSessions = Array.from(this.sessions.values()); + const activeSessions = allSessions.filter(s => s.status === 'running'); + const completedSessions = allSessions.filter(s => s.status === 'completed'); + const failedSessions = allSessions.filter(s => s.status === 'failed'); + + const totalComputeSeconds = allSessions.reduce( + (sum, s) => sum + (s.durationSeconds || 0), + 0 + ); + + const totalCost = allSessions.reduce( + (sum, s) => sum + (s.costUSD || 0), + 0 + ); + + const completedDurations = completedSessions + .map(s => s.durationSeconds || 0) + .filter(d => d > 0); + + const averageDuration = completedDurations.length > 0 + ? completedDurations.reduce((a, b) => a + b, 0) / completedDurations.length + : 0; + + // Calculate peak concurrent agents from events + const peakConcurrent = this.calculatePeakConcurrency(); + + return { + totalAgents: allSessions.length, + activeAgents: activeSessions.length, + completedTasks: completedSessions.length, + failedTasks: failedSessions.length, + totalComputeHours: totalComputeSeconds / 3600, + totalCostUSD: totalCost, + averageTaskDuration: averageDuration, + peakConcurrentAgents: peakConcurrent + }; + } + + /** + * Calculate peak concurrent agents from event history + */ + private calculatePeakConcurrency(): number { + const timePoints: Array<{ time: number; delta: number }> = []; + + for (const session of this.sessions.values()) { + timePoints.push({ time: session.startTime, delta: 1 }); + if (session.endTime) { + timePoints.push({ time: session.endTime, delta: -1 }); + } + } + + timePoints.sort((a, b) => a.time - b.time); + + let current = 0; + let peak = 0; + + for (const point of timePoints) { + current += point.delta; + peak = Math.max(peak, current); + } + + return peak; + } + + /** + * Check if budget alerts should be triggered + */ + private checkBudgetAlerts(): void { + if (!this.costConfig.budgetLimitUSD) return; + + const metrics = this.getMetrics(); + const percentageUsed = (metrics.totalCostUSD / this.costConfig.budgetLimitUSD) * 100; + + for (const threshold of this.costConfig.alertThresholds) { + if (percentageUsed >= threshold && !this.alertedThresholds.has(threshold)) { + this.alertedThresholds.add(threshold); + + const severity = threshold >= 90 ? 'critical' : + threshold >= 75 ? 'warning' : 'info'; + + const alert: BudgetAlert = { + timestamp: Date.now(), + currentSpend: metrics.totalCostUSD, + budgetLimit: this.costConfig.budgetLimitUSD, + percentageUsed, + message: `Budget alert: ${percentageUsed.toFixed(1)}% of budget used ($${metrics.totalCostUSD.toFixed(2)} / $${this.costConfig.budgetLimitUSD})`, + severity + }; + + this.alerts.push(alert); + } + } + } + + /** + * Get all budget alerts + */ + getAlerts(): BudgetAlert[] { + return [...this.alerts]; + } + + /** + * Get recent alerts (last N) + */ + getRecentAlerts(count: number = 10): BudgetAlert[] { + return this.alerts.slice(-count); + } + + /** + * Clear all alerts + */ + clearAlerts(): void { + this.alerts = []; + this.alertedThresholds.clear(); + } + + /** + * Get usage breakdown by task + */ + getUsageByTask(): Map { + const taskMap = new Map(); + + for (const session of this.sessions.values()) { + const existing = taskMap.get(session.taskId) || { agents: 0, cost: 0, duration: 0 }; + + taskMap.set(session.taskId, { + agents: existing.agents + 1, + cost: existing.cost + (session.costUSD || 0), + duration: existing.duration + (session.durationSeconds || 0) + }); + } + + return taskMap; + } + + /** + * Get usage breakdown by time period + */ + getUsageByPeriod(periodMs: number): Array<{ period: number; agents: number; cost: number }> { + const periods = new Map; cost: number }>(); + + for (const session of this.sessions.values()) { + const periodStart = Math.floor(session.startTime / periodMs) * periodMs; + + const existing = periods.get(periodStart) || { agents: new Set(), cost: 0 }; + existing.agents.add(session.agentId); + existing.cost += session.costUSD || 0; + + periods.set(periodStart, existing); + } + + return Array.from(periods.entries()) + .map(([period, data]) => ({ + period, + agents: data.agents.size, + cost: data.cost + })) + .sort((a, b) => a.period - b.period); + } + + /** + * Export usage data for analysis + */ + exportData(): { + sessions: AgentSession[]; + events: AgentUsageEvent[]; + metrics: UsageMetrics; + alerts: BudgetAlert[]; + } { + return { + sessions: Array.from(this.sessions.values()), + events: [...this.events], + metrics: this.getMetrics(), + alerts: [...this.alerts] + }; + } + + /** + * Reset all tracking data + */ + reset(): void { + this.sessions.clear(); + this.events = []; + this.alerts = []; + this.alertedThresholds.clear(); + } +} + +/** + * Default cost configuration (example AWS pricing) + */ +export const DEFAULT_COST_CONFIG: CostConfig = { + cpuCostPerCoreHour: 0.04, // ~$0.04 per vCPU hour + memoryCostPerGBHour: 0.005, // ~$0.005 per GB hour + budgetLimitUSD: 100, // $100 default budget + alertThresholds: [50, 75, 90, 95] // Alert at 50%, 75%, 90%, 95% +}; diff --git a/desktop-client/src/api/orchestrator-client.ts b/desktop-client/src/api/orchestrator-client.ts new file mode 100644 index 0000000..769e292 --- /dev/null +++ b/desktop-client/src/api/orchestrator-client.ts @@ -0,0 +1,215 @@ +/** + * REST API client for the orchestrator service + */ + +export interface AgentMetadata { + agent_id: string; + status: 'idle' | 'busy' | 'handoff' | 'handoff-pending' | 'failed'; + capabilities: string[]; + current_task_id?: string; + last_heartbeat: number; + created_at: number; +} + +export interface TaskMetadata { + task_id: string; + title?: string; + description: string; + status: 'pending' | 'assigned' | 'in_progress' | 'blocked' | 'completed' | 'failed' | 'cancelled'; + agent_role?: string; + required_capabilities?: string[]; + depends_on?: string[]; + parent_task_id?: string; + root_task_id?: string; + source?: string; + assigned_agent_id?: string; + context: Record; + created_at: number; + started_at?: number; + completed_at?: number; + attempt?: number; + blocked_reason?: string; + max_retries: number; + retry_count: number; + error?: string; +} + +export interface HandoffMetadata { + handoff_id: string; + source_agent_id: string; + target_agent_id: string; + task_context: Record; + status: 'pending' | 'completed' | 'failed'; + created_at: number; + completed_at?: number; + reason?: string; +} + +export interface HealthStatus { + status: string; + redis: string; + active_connections: number; +} + +export interface OrchestratorClientConfig { + baseUrl: string; + timeout?: number; +} + +export class OrchestratorClient { + private baseUrl: string; + private timeout: number; + + constructor(config: OrchestratorClientConfig) { + this.baseUrl = config.baseUrl.replace(/\/$/, ''); // Remove trailing slash + this.timeout = config.timeout || 10000; + } + + /** + * Make HTTP request with timeout + */ + private async request( + endpoint: string, + options: RequestInit = {} + ): Promise { + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), this.timeout); + + try { + const response = await fetch(`${this.baseUrl}${endpoint}`, { + ...options, + signal: controller.signal, + headers: { + 'Content-Type': 'application/json', + ...options.headers, + }, + }); + + if (!response.ok) { + const error = await response.text(); + throw new Error(`HTTP ${response.status}: ${error}`); + } + + return await response.json(); + } finally { + clearTimeout(timeoutId); + } + } + + /** + * Health check + */ + async health(): Promise { + return this.request('/health'); + } + + /** + * List all agents + */ + async listAgents(): Promise { + const response = await this.request<{ agents: AgentMetadata[] }>('/agents'); + return response.agents; + } + + /** + * Get specific agent + */ + async getAgent(agentId: string): Promise { + return this.request(`/agents/${agentId}`); + } + + /** + * List idle agents + */ + async listIdleAgents(): Promise { + const response = await this.request<{ agents: AgentMetadata[] }>('/agents/idle'); + return response.agents; + } + + /** + * Create a new task + */ + async createTask( + description: string, + context: Record = {}, + maxRetries: number = 3 + ): Promise { + return this.request('/tasks', { + method: 'POST', + body: JSON.stringify({ + description, + context, + max_retries: maxRetries, + }), + }); + } + + /** + * Assign task to agent + */ + async assignTask(taskId: string, agentId: string): Promise { + await this.request('/tasks/assign', { + method: 'POST', + body: JSON.stringify({ + task_id: taskId, + agent_id: agentId, + }), + }); + } + + /** + * List all tasks + */ + async listTasks(status?: string): Promise { + const query = status ? `?status=${status}` : ''; + const response = await this.request<{ tasks: TaskMetadata[] }>(`/tasks${query}`); + return response.tasks; + } + + /** + * Get specific task + */ + async getTask(taskId: string): Promise { + return this.request(`/tasks/${taskId}`); + } + + /** + * List handoff history + */ + async listHandoffs(agentId?: string, limit: number = 100): Promise { + const params = new URLSearchParams(); + if (agentId) params.append('agent_id', agentId); + params.append('limit', limit.toString()); + + const query = params.toString() ? `?${params.toString()}` : ''; + const response = await this.request<{ handoffs: HandoffMetadata[] }>(`/handoffs${query}`); + return response.handoffs; + } + + /** + * Create WebSocket connection for real-time updates + */ + createWebSocket(agentId: string): WebSocket { + const wsUrl = this.baseUrl.replace(/^http/, 'ws'); + return new WebSocket(`${wsUrl}/ws/${agentId}`); + } +} + +/** + * Singleton instance + */ +let clientInstance: OrchestratorClient | null = null; + +export function getOrchestratorClient(config?: OrchestratorClientConfig): OrchestratorClient { + if (!clientInstance) { + if (!config) { + throw new Error('OrchestratorClient not initialized. Provide config on first call.'); + } + clientInstance = new OrchestratorClient(config); + } + return clientInstance; +} + +export function resetOrchestratorClient(): void { + clientInstance = null; +} diff --git a/desktop-client/src/components/AgentDashboard.tsx b/desktop-client/src/components/AgentDashboard.tsx new file mode 100644 index 0000000..1db32f2 --- /dev/null +++ b/desktop-client/src/components/AgentDashboard.tsx @@ -0,0 +1,225 @@ +import React, { useEffect, useState } from 'react'; +import { AgentMetadata, getOrchestratorClient } from '../api/orchestrator-client'; + +export interface AgentDashboardProps { + orchestratorUrl: string; + workspaceId?: string; +} + +export const AgentDashboard: React.FC = ({ + orchestratorUrl, + workspaceId, +}) => { + const [agents, setAgents] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [connected, setConnected] = useState(false); + + useEffect(() => { + const client = getOrchestratorClient({ baseUrl: orchestratorUrl }); + + // Initial fetch + const fetchAgents = async () => { + try { + const agentList = await client.listAgents(); + setAgents(agentList); + setConnected(true); + setError(null); + } catch (err) { + setError(err instanceof Error ? err.message : 'Failed to fetch agents'); + setConnected(false); + } finally { + setLoading(false); + } + }; + + fetchAgents(); + + // Poll for updates every 2 seconds + const interval = setInterval(fetchAgents, 2000); + + return () => clearInterval(interval); + }, [orchestratorUrl]); + + const getStatusColor = (status: string) => { + switch (status) { + case 'idle': + return 'bg-gray-500'; + case 'busy': + return 'bg-blue-500 animate-pulse'; + case 'handoff': + case 'handoff-pending': + return 'bg-yellow-500 animate-pulse'; + case 'failed': + return 'bg-red-500'; + default: + return 'bg-gray-500'; + } + }; + + const getStatusText = (status: string) => { + switch (status) { + case 'idle': + return 'Idle'; + case 'busy': + return 'Working'; + case 'handoff': + case 'handoff-pending': + return 'Handoff'; + case 'failed': + return 'Failed'; + default: + return 'Unknown'; + } + }; + + const formatTimestamp = (timestamp: number) => { + const date = new Date(timestamp * 1000); + const now = Date.now(); + const diff = now - date.getTime(); + + if (diff < 60000) { + return 'Just now'; + } else if (diff < 3600000) { + return `${Math.floor(diff / 60000)}m ago`; + } else if (diff < 86400000) { + return `${Math.floor(diff / 3600000)}h ago`; + } else { + return date.toLocaleDateString(); + } + }; + + const filteredAgents = workspaceId + ? agents.filter(agent => agent.agent_id.includes(workspaceId)) + : agents; + + if (loading) { + return ( +
+

Agent Dashboard

+
+ + + + + Loading agents... +
+
+ ); + } + + return ( +
+
+

Agent Dashboard

+
+
+ + {connected ? 'Connected' : 'Disconnected'} + +
+
+ + {error && ( +
+ {error} +
+ )} + +
+
+

Total Agents

+

{filteredAgents.length}

+
+
+

Active

+

+ {filteredAgents.filter(a => a.status === 'busy').length} +

+
+
+

Idle

+

+ {filteredAgents.filter(a => a.status === 'idle').length} +

+
+
+ + {filteredAgents.length === 0 ? ( +
+

No agents running

+

Submit a task to create agents

+
+ ) : ( +
+ {filteredAgents.map((agent) => ( +
+
+
+
+
+ + {agent.agent_id} + +
+ +
+
+ Status: + {getStatusText(agent.status)} +
+ + {agent.current_task_id && ( +
+ Task: + + {agent.current_task_id} + +
+ )} + +
+ Last Heartbeat: + {formatTimestamp(agent.last_heartbeat)} +
+ + {agent.capabilities.length > 0 && ( +
+ Capabilities: +
+ {agent.capabilities.map((cap, idx) => ( + + {cap} + + ))} +
+
+ )} +
+
+
+
+ ))} +
+ )} +
+ ); +}; diff --git a/desktop-client/src/components/CostControlDialog.tsx b/desktop-client/src/components/CostControlDialog.tsx new file mode 100644 index 0000000..bac7611 --- /dev/null +++ b/desktop-client/src/components/CostControlDialog.tsx @@ -0,0 +1,260 @@ +import React, { useState } from 'react'; +import { ComplexityAnalysis } from '../ai/complexity-analyzer'; + +export interface CostControlDialogProps { + analysis: ComplexityAnalysis; + maxAgents: number; + onConfirm: () => void; + onCancel: () => void; + onAdjust: (newCount: number) => void; +} + +export const CostControlDialog: React.FC = ({ + analysis, + maxAgents, + onConfirm, + onCancel, + onAdjust, +}) => { + const [adjustedCount, setAdjustedCount] = useState(analysis.agentCount); + const agentCount = analysis.agentCount; + + // Determine warning level + const isWarning = agentCount > 10; + const requiresConfirmation = agentCount > 15; + const exceedsMax = agentCount > maxAgents; + + // Calculate estimated resource cost + const estimatedCpuHours = (agentCount * 0.5).toFixed(1); // Assume 30min avg per agent + const estimatedCost = (agentCount * 0.05).toFixed(2); // Rough cost estimate + + if (!isWarning && !exceedsMax) { + // No warning needed, auto-confirm + return null; + } + + return ( +
+
+ {/* Header */} +
+
+ + + +
+
+

+ {exceedsMax + ? 'Agent Limit Exceeded' + : requiresConfirmation + ? 'High Agent Count Confirmation Required' + : 'High Agent Count Warning'} +

+

+ {exceedsMax + ? `The task requires ${agentCount} agents, which exceeds your configured maximum of ${maxAgents}.` + : requiresConfirmation + ? `This task will create ${agentCount} agents. Please confirm this is intentional.` + : `This task will create ${agentCount} agents, which may consume significant resources.`} +

+
+
+ + {/* Analysis Details */} +
+

Task Analysis

+
+
+ Complexity: + {analysis.complexity} +
+
+ Recommended Agents: + {agentCount} +
+
+ Confidence: + {(analysis.confidence * 100).toFixed(0)}% +
+
+
+

+ Reasoning: {analysis.reasoning} +

+
+
+ + {/* Resource Estimates */} +
+

Estimated Resource Usage

+
+
+ CPU Hours: + ~{estimatedCpuHours}h +
+
+ Estimated Cost: + ${estimatedCost} +
+
+
+ + {/* Agent Count Adjustment */} + {!exceedsMax && ( +
+ +
+ setAdjustedCount(parseInt(e.target.value, 10))} + className="flex-1" + /> + setAdjustedCount(parseInt(e.target.value, 10))} + className="w-20 px-3 py-2 border border-gray-300 rounded-md text-sm" + /> +
+ {adjustedCount !== agentCount && ( +

+ Adjusting from {agentCount} to {adjustedCount} agents may affect task completion + quality. +

+ )} +
+ )} + + {/* Actions */} +
+ + {exceedsMax ? ( + + ) : ( + <> + {adjustedCount !== agentCount && ( + + )} + + + )} +
+
+
+ ); +}; + +/** + * Hook to manage cost control dialog state + */ +export function useCostControl(maxAgents: number = 20) { + const [showDialog, setShowDialog] = useState(false); + const [currentAnalysis, setCurrentAnalysis] = useState(null); + const [onConfirmCallback, setOnConfirmCallback] = useState<(() => void) | null>(null); + + const checkAndConfirm = ( + analysis: ComplexityAnalysis, + onConfirm: () => void + ): Promise => { + return new Promise((resolve) => { + const agentCount = analysis.agentCount; + + // Auto-approve if within safe limits + if (agentCount <= 10 && agentCount <= maxAgents) { + onConfirm(); + resolve(true); + return; + } + + // Show dialog for high counts or exceeding max + setCurrentAnalysis(analysis); + setOnConfirmCallback(() => () => { + onConfirm(); + setShowDialog(false); + resolve(true); + }); + setShowDialog(true); + }); + }; + + const handleCancel = () => { + setShowDialog(false); + setCurrentAnalysis(null); + setOnConfirmCallback(null); + }; + + const handleAdjust = (newCount: number) => { + if (currentAnalysis) { + const adjustedAnalysis = { ...currentAnalysis, agentCount: newCount }; + setCurrentAnalysis(adjustedAnalysis); + } + }; + + return { + showDialog, + currentAnalysis, + checkAndConfirm, + handleCancel, + handleConfirm: onConfirmCallback, + handleAdjust, + }; +} diff --git a/desktop-client/src/components/ResultViewer.tsx b/desktop-client/src/components/ResultViewer.tsx new file mode 100644 index 0000000..653c1e7 --- /dev/null +++ b/desktop-client/src/components/ResultViewer.tsx @@ -0,0 +1,238 @@ +import React, { useState } from 'react'; +import { MergeConflict, ConflictResolution } from '../git/merge-manager'; + +export interface ResultViewerProps { + conflicts: MergeConflict[]; + mergedBranches: string[]; + onResolve: (resolutions: ConflictResolution[]) => void; + onCancel: () => void; +} + +export const ResultViewer: React.FC = ({ + conflicts, + mergedBranches, + onResolve, + onCancel, +}) => { + const [resolutions, setResolutions] = useState>(new Map()); + const [selectedFile, setSelectedFile] = useState( + conflicts.length > 0 ? conflicts[0].filepath : null + ); + + const handleResolutionChange = ( + filepath: string, + resolution: 'ours' | 'theirs' | 'manual', + manualContent?: string + ) => { + const newResolutions = new Map(resolutions); + newResolutions.set(filepath, { + filepath, + resolution, + manualContent, + }); + setResolutions(newResolutions); + }; + + const handleResolveAll = () => { + const allResolutions = Array.from(resolutions.values()); + + // Ensure all conflicts have resolutions + const unresolvedConflicts = conflicts.filter( + conflict => !resolutions.has(conflict.filepath) + ); + + if (unresolvedConflicts.length > 0) { + alert(`Please resolve all conflicts. ${unresolvedConflicts.length} remaining.`); + return; + } + + onResolve(allResolutions); + }; + + const selectedConflict = conflicts.find(c => c.filepath === selectedFile); + const currentResolution = selectedFile ? resolutions.get(selectedFile) : null; + + if (conflicts.length === 0) { + return ( +
+

Merge Results

+ +
+
+ + + + Merge completed successfully! +
+
+ +
+

Merged branches:

+
+ {mergedBranches.map((branch, idx) => ( +
+ + + + {branch} +
+ ))} +
+
+
+ ); + } + + return ( +
+

Resolve Merge Conflicts

+ +
+
+ + + + + {conflicts.length} conflict{conflicts.length > 1 ? 's' : ''} detected + +
+
+ +
+ {/* File list */} +
+

Conflicted Files

+
+ {conflicts.map((conflict) => { + const resolved = resolutions.has(conflict.filepath); + return ( + + ); + })} +
+
+ + {/* Conflict viewer and resolution */} +
+ {selectedConflict && ( + <> +
+

+ {selectedConflict.filepath} +

+ +
+ + + + +
+
+ Manual Resolution + {currentResolution?.resolution === 'manual' && ( + + + + )} +
+