diff --git a/RELEASE_NOTES_v1.3.0.md b/RELEASE_NOTES_v1.3.0.md new file mode 100644 index 0000000..f5f94ef --- /dev/null +++ b/RELEASE_NOTES_v1.3.0.md @@ -0,0 +1,78 @@ +# AgentFM v1.3.0 — Verifiable Agent Mesh + +AgentFM is a peer-to-peer compute mesh for running containerized AI agents on idle hardware. **v1.3.0 makes the mesh trustworthy:** it adds a tamper-evident reputation system so dishonest agents get caught and ejected — with no blockchain, tokens, or staking anywhere in it. + +This is the biggest release so far (276 commits since 1.2.0). Here's what's in it. + +## Trust you can actually verify + +Every rating and comment an agent earns now lives in a **signed, append-only ledger** — one per peer, built on a Merkle log. Nothing can be edited or deleted after the fact, and anyone can independently check a peer's history with cryptographic inclusion proofs. No central server, no chain. + +- **Signed feedback.** Ratings and comments are Ed25519-signed and gossiped across the mesh on a dedicated feedback topic. Replays, forged signatures, and out-of-range scores are rejected. +- **Witnesses catch liars.** Peers can act as witnesses that co-sign each other's ledger heads. If an agent shows a different history to different peers ("equivocation"), the witnesses detect the fork and the offender is **permanently pinned to the lowest score, -1.0**. +- **Reputation that means something.** Scoring uses an EigenTrust-style model: your vote weighs as much as your own reputation, recent feedback counts more than old, and agents that drop below the threshold are auto-ejected (with hysteresis so they don't flap in and out). + +## Know what you're running + +- Agents now advertise their **container image digest and capability** in telemetry. +- At dispatch, the Boss checks that image against a curated **trusted-agents registry** in one of three modes — `off`, `warn`, or `strict`. Strict refuses anything unrecognized; a mismatch costs the agent reputation. +- Known equivocators are always refused, regardless of mode. + +## A new desktop app + +A full **desktop app (the "Boss")**, built with Electron and React, ships alongside the CLI. Discover agents on a live mesh radar, dispatch tasks and watch the output stream in, browse the artifacts they produce, leave signed ratings, and inspect any peer's reputation ledger — all without touching the terminal. The interactive TUI is still there if you prefer it. + +## API, SDK, and web viewer + +- New reputation endpoints on the Boss HTTP gateway: read a peer's reputation, its full log, an inclusion proof for any entry, and submit signed comments. +- The **Python SDK** gains a `client.reputation` namespace (`get` / `log` / `proof` / `comment`). +- A built-in **web viewer** shows any peer's reputation history at `/ui/peer/{id}`, auto-refreshing. + +## One binary, every role + +The standalone relay binary is **gone** — it's folded into the main binary. Every role now runs from the same `agentfm` executable: + +```sh +agentfm # Boss: dispatch tasks + HTTP gateway +agentfm -mode worker # host agents in Podman sandboxes +agentfm -mode relay # public lighthouse / circuit relay +agentfm --witness # co-sign the reputation ledger +``` + +## Security hardening + +Fixes from an internal audit, landed before release: + +- Rogue witnesses can no longer forge equivocation alerts against innocent peers. +- Witness extension checks now require a valid consistency proof, closing a fork-at-mismatched-size hole. +- Post-signature validation rejects NaN / infinite / out-of-range scores. +- All protocol framing has hard size caps (witness frames, comment bodies, fetch batches, image cache). + +## Install + +```sh +curl -fsSL https://api.agentfm.net/install.sh | bash +``` + +Or grab a binary for your platform from the release assets and `chmod +x` it. They're static, with no dependencies: + +| OS | Architectures | +|----|----------------| +| macOS | `darwin_arm64` (Apple Silicon), `darwin_amd64` (Intel) | +| Linux | `amd64`, `arm64`, `arm` (v7), `386`, `riscv64` | +| Windows | `amd64`, `arm64` | +| FreeBSD | `amd64` | + +Verify your download against `checksums.txt` (SHA-256). + +## Coming later + +A few pieces are intentionally deferred: salt-challenge image probes and the real Sigstore Rekor client (v1.3.1); golden-prompt probes, an LLM-judge grader, delegated comments, and an optional TEE attestation tier (v1.4). + +## On purpose: no blockchain + +AgentFM has no blockchain, no token, no staking, and no on-chain governance, and it never will. Trust here comes from signed logs and witnesses — not consensus or coins. + +--- + +Full technical changelog: [`CHANGELOG.md`](./CHANGELOG.md) diff --git a/agentfm-desktop/electron/ipc.ts b/agentfm-desktop/electron/ipc.ts index f73b8db..391b0b7 100644 --- a/agentfm-desktop/electron/ipc.ts +++ b/agentfm-desktop/electron/ipc.ts @@ -21,6 +21,19 @@ const ALLOWED_SETTINGS_KEYS = new Set([ 'activeProjectId', ]) +// Chat sessions persist under a per-project key (chat:sessions:), +// so they can't live in the static allowlist. Permit exactly that shape with +// a conservative project-id suffix (matches the ids minted in projectStore). +const CHAT_SESSIONS_KEY = /^chat:sessions:[A-Za-z0-9_-]{1,64}$/ + +// isAllowedSettingsKey gates every renderer settings:set. A key passes when it +// is a dot-free member of the static allowlist OR a well-formed per-project +// chat-sessions key. Exported for unit testing. +export function isAllowedSettingsKey(key: unknown): key is string { + if (typeof key !== 'string' || key.includes('.')) return false + return ALLOWED_SETTINGS_KEYS.has(key) || CHAT_SESSIONS_KEY.test(key) +} + const SWARM_KEY_HEX = /^[0-9a-fA-F]{64}$/ // validateProjects checks the security-relevant fields of each stored project @@ -78,7 +91,7 @@ export function registerIPC(backend: BackendManager): void { // Persistent settings ipcMain.handle('settings:get', (_event, key: string) => settingsStore.get(key as never)) ipcMain.handle('settings:set', (_event, key: string, value: unknown) => { - if (typeof key !== 'string' || key.includes('.') || !ALLOWED_SETTINGS_KEYS.has(key)) { + if (!isAllowedSettingsKey(key)) { throw new Error(`refused settings key: ${JSON.stringify(key)}`) } let safeValue = value diff --git a/agentfm-desktop/electron/main.ts b/agentfm-desktop/electron/main.ts index 1ae4fb4..ff5b476 100644 --- a/agentfm-desktop/electron/main.ts +++ b/agentfm-desktop/electron/main.ts @@ -6,6 +6,11 @@ import { BackendManager } from './backend-manager' import { registerIPC, isSafeExternalUrl } from './ipc' import { settingsStore } from './store' +// Set before app 'ready' so the macOS menu bar, app switcher, and dock use +// "AgentFM" instead of the default "Electron" in dev. The packaged app gets +// this from electron-builder.yml (productName), but dev runs the raw bundle. +app.setName('AgentFM') + let backend: BackendManager | null = null function resolveAppIcon(): Electron.NativeImage | undefined { @@ -30,6 +35,7 @@ function createWindow(): void { minWidth: 900, minHeight: 600, show: false, + title: 'AgentFM', icon: appIcon, autoHideMenuBar: true, titleBarStyle: process.platform === 'darwin' ? 'hiddenInset' : 'default', diff --git a/agentfm-desktop/package.json b/agentfm-desktop/package.json index 79a1127..a71c6ba 100644 --- a/agentfm-desktop/package.json +++ b/agentfm-desktop/package.json @@ -7,6 +7,7 @@ "license": "MIT", "main": "out/main/index.js", "scripts": { + "predev": "node scripts/brand-dev-bundle.mjs", "dev": "electron-vite dev", "build": "electron-vite build", "start": "electron-vite preview", diff --git a/agentfm-desktop/scripts/brand-dev-bundle.mjs b/agentfm-desktop/scripts/brand-dev-bundle.mjs new file mode 100644 index 0000000..5d2b435 --- /dev/null +++ b/agentfm-desktop/scripts/brand-dev-bundle.mjs @@ -0,0 +1,28 @@ +// Renames the dev Electron.app bundle to "AgentFM" so the macOS Dock and menu +// show the product name during `npm run dev`. The Dock reads CFBundleName from +// the running .app bundle, which app.setName() cannot override in dev — only +// this can. No-op on non-macOS or when the bundle is absent (CI / fresh clone +// before electron is installed). Packaged builds get their name from +// electron-builder.yml (productName) and never hit this path. +import { execFileSync } from 'node:child_process' +import { existsSync } from 'node:fs' + +if (process.platform !== 'darwin') process.exit(0) + +const plist = 'node_modules/electron/dist/Electron.app/Contents/Info.plist' +if (!existsSync(plist)) process.exit(0) + +function setKey(key, value) { + try { + execFileSync('/usr/libexec/PlistBuddy', ['-c', `Set :${key} ${value}`, plist]) + } catch { + try { + execFileSync('/usr/libexec/PlistBuddy', ['-c', `Add :${key} string ${value}`, plist]) + } catch { + // best-effort dev convenience; never fail the build over it + } + } +} + +setKey('CFBundleName', 'AgentFM') +setKey('CFBundleDisplayName', 'AgentFM') diff --git a/agentfm-desktop/src/components/peer/EntryRow.tsx b/agentfm-desktop/src/components/peer/EntryRow.tsx index ea0b13d..366a363 100644 --- a/agentfm-desktop/src/components/peer/EntryRow.tsx +++ b/agentfm-desktop/src/components/peer/EntryRow.tsx @@ -1,5 +1,7 @@ +import { Copy } from 'lucide-react'; +import { toast } from 'sonner'; import type { PeerEntry } from '../../types/api'; -import { compactAge } from '../../lib/peer'; +import { compactAge, shortenPeerID } from '../../lib/peer'; import { CommentBody } from './CommentBody'; import { StarRow } from '../primitives/StarRow'; import { starsFromScore } from '../../lib/stars'; @@ -11,8 +13,6 @@ interface Props { export function EntryRow({ entry, peerId }: Props) { const isComment = entry.kind === 'Comment'; - const unverified = entry.rater_status === 'unverified'; - const hasMeta = unverified || !!entry.context; return (
@@ -44,16 +44,24 @@ export function EntryRow({ entry, peerId }: Props) { ) : null}
- {hasMeta && ( -
- {unverified && ( - - unverified - - )} - {entry.context && {entry.context}} -
- )} +
+ {entry.rater_peer_id && ( + + )} + {entry.context && {entry.context}} +
{isComment && entry.text_cid && }
diff --git a/agentfm-desktop/src/components/peer/SummaryCard.tsx b/agentfm-desktop/src/components/peer/SummaryCard.tsx index 61068e0..00e47ba 100644 --- a/agentfm-desktop/src/components/peer/SummaryCard.tsx +++ b/agentfm-desktop/src/components/peer/SummaryCard.tsx @@ -84,13 +84,6 @@ export function SummaryCard({ data }: { data: PeerSummary }) { {data.entries_count} - - - {data.rater_summary?.verified_raters_count ?? 0} verified - - ); } diff --git a/agentfm-desktop/src/lib/query.ts b/agentfm-desktop/src/lib/query.ts index 93b168f..b1bcebf 100644 --- a/agentfm-desktop/src/lib/query.ts +++ b/agentfm-desktop/src/lib/query.ts @@ -20,7 +20,15 @@ export const qk = { } export function useAbout() { - return useQuery({ queryKey: qk.about(), queryFn: api.about, staleTime: 5000 }) + // Poll: relay connectedness, ledger tree size, and uptime are all live. + // Without an interval the toolbar would latch onto the boot-time snapshot + // (relay not yet dialed) and show "Connecting to relay…" forever. + return useQuery({ + queryKey: qk.about(), + queryFn: api.about, + staleTime: 5000, + refetchInterval: 5000, + }) } export function useWorkers(includeOffline = false) { diff --git a/agentfm-desktop/src/routes/Status.tsx b/agentfm-desktop/src/routes/Status.tsx index 6e4cbf4..f2b208b 100644 --- a/agentfm-desktop/src/routes/Status.tsx +++ b/agentfm-desktop/src/routes/Status.tsx @@ -32,7 +32,7 @@ function StatCard({ label, icon: Icon, value, valueTone, sub }: StatCardProps) { ok: COLORS.ok, accent: COLORS.accent, bad: COLORS.bad, }[valueTone] return ( - +
{label}
@@ -72,7 +72,6 @@ export default function Status() { {/* Health banner */}
{onlineWorkers} agent{onlineWorkers === 1 ? '' : 's'} online - {isPrivate ? 'private swarm' : 'public mesh'} - up {formatUptime(uptime)}
diff --git a/agentfm-desktop/tests/unit/ipcGuards.test.ts b/agentfm-desktop/tests/unit/ipcGuards.test.ts index 006df9e..75aa8cc 100644 --- a/agentfm-desktop/tests/unit/ipcGuards.test.ts +++ b/agentfm-desktop/tests/unit/ipcGuards.test.ts @@ -10,7 +10,7 @@ vi.mock('electron', () => ({ })) vi.mock('../../electron/store', () => ({ settingsStore: { get: vi.fn(), set: vi.fn(), delete: vi.fn() } })) -import { isSafeExternalUrl } from '../../electron/ipc' +import { isSafeExternalUrl, isAllowedSettingsKey } from '../../electron/ipc' describe('isSafeExternalUrl', () => { it('allows web and mail links', () => { @@ -26,3 +26,26 @@ describe('isSafeExternalUrl', () => { expect(isSafeExternalUrl('not a url')).toBe(false) }) }) + +describe('isAllowedSettingsKey', () => { + it('allows the static settings keys', () => { + expect(isAllowedSettingsKey('theme')).toBe(true) + expect(isAllowedSettingsKey('projects')).toBe(true) + expect(isAllowedSettingsKey('activeProjectId')).toBe(true) + }) + + it('allows per-project chat-session keys', () => { + expect(isAllowedSettingsKey('chat:sessions:prj_default')).toBe(true) + expect(isAllowedSettingsKey('chat:sessions:prj_private_ag')).toBe(true) + expect(isAllowedSettingsKey('chat:sessions:prj_h4af90vy')).toBe(true) + }) + + it('refuses malformed or unlisted keys', () => { + expect(isAllowedSettingsKey('apiPort.x')).toBe(false) + expect(isAllowedSettingsKey('chat:sessions:')).toBe(false) + expect(isAllowedSettingsKey('chat:sessions:bad id!')).toBe(false) + expect(isAllowedSettingsKey('chat:sessions:a.b')).toBe(false) + expect(isAllowedSettingsKey('arbitrary')).toBe(false) + expect(isAllowedSettingsKey(42 as unknown)).toBe(false) + }) +}) diff --git a/agentfm-go/cmd/agentfm/bossbootstrap.go b/agentfm-go/cmd/agentfm/bossbootstrap.go index 619717f..cd6616d 100644 --- a/agentfm-go/cmd/agentfm/bossbootstrap.go +++ b/agentfm-go/cmd/agentfm/bossbootstrap.go @@ -81,8 +81,15 @@ func bossOptionsFromFlags( return opts, cleanup } + cstore, err := comments.Open(defaultCommentsRoot()) + if err != nil { + slog.Warn("boss bootstrap: comments store open failed; P4-3 disabled", + slog.Any(obs.FieldErr, err)) + } + l, err := ledger.NewWithOptions(dbPath, priv, node.PubSub, ledger.Options{ - Host: node.Host, + Host: node.Host, + Comments: cstore, }) if err != nil { slog.Warn("boss bootstrap: ledger open failed; v1.3 endpoints will 503", @@ -203,11 +210,9 @@ func bossOptionsFromFlags( } // --- comments store + submission handler -------------------------- - cstore, err := comments.Open(defaultCommentsRoot()) - if err != nil { - slog.Warn("boss bootstrap: comments store open failed; P4-3 disabled", - slog.Any(obs.FieldErr, err)) - } else { + // cstore was opened before the ledger so Options.Comments could be + // wired; nil here means the open failed and P4-3 stays disabled. + if cstore != nil { cserver := comments.NewServer(node.Host, cstore) cserver.Start() cleanups = append(cleanups, cserver.Stop) diff --git a/agentfm-go/cmd/agentfm/relay.go b/agentfm-go/cmd/agentfm/relay.go index c15be22..eb90730 100644 --- a/agentfm-go/cmd/agentfm/relay.go +++ b/agentfm-go/cmd/agentfm/relay.go @@ -10,6 +10,7 @@ import ( "syscall" "agentfm/internal/ledger" + "agentfm/internal/ledger/comments" "agentfm/internal/metrics" "agentfm/internal/network" "agentfm/internal/obs" @@ -21,8 +22,9 @@ import ( // infinite reservation limits, a Kademlia DHT in server mode, an actively // drained telemetry subscription (so it keeps routing gossip), and a full // archive ledger. The archive persists every signed Rating / Comment / -// EquivocationAlert and serves head-fetch / ledger-fetch, so a fresh boss -// can catch up against this relay even when every other boss is offline. +// EquivocationAlert, serves head-fetch / ledger-fetch, and replicates + +// serves comment bodies (comment-fetch), so a fresh boss can catch up +// against this relay even when every other boss is offline. // // This is the single relay path — the dedicated relay binary was folded in // here so `agentfm -mode relay` is the only relay, dev or production. @@ -73,11 +75,22 @@ func runRelayMode(ctx context.Context, netCfg network.Config, promListen string) pterm.Fatal.Printfln("relay identity load failed: %v", err) } + var cstore *comments.Store + if cs, err := comments.Open(defaultCommentsRoot()); err != nil { + slog.Warn("relay: comments store open failed; comment bodies will not be archived", + slog.Any(obs.FieldErr, err)) + } else { + cstore = cs + cserver := comments.NewServer(node.Host, cstore) + cserver.Start() + defer cserver.Stop() + } + dbPath := defaultRelayLedgerPath() if err := os.MkdirAll(filepath.Dir(dbPath), 0o700); err != nil { slog.Warn("relay: could not create ledger dir; running connectivity-only", slog.String("path", filepath.Dir(dbPath)), slog.Any(obs.FieldErr, err)) - } else if arch, err := ledger.NewWithOptions(dbPath, priv, node.PubSub, ledger.Options{Host: node.Host}); err != nil { + } else if arch, err := ledger.NewWithOptions(dbPath, priv, node.PubSub, ledger.Options{Host: node.Host, Comments: cstore}); err != nil { slog.Warn("relay: archive ledger failed to open; running connectivity-only", slog.Any(obs.FieldErr, err)) } else { diff --git a/agentfm-go/cmd/agentfm/witness.go b/agentfm-go/cmd/agentfm/witness.go index a74f769..7e3498a 100644 --- a/agentfm-go/cmd/agentfm/witness.go +++ b/agentfm-go/cmd/agentfm/witness.go @@ -57,8 +57,15 @@ func runWitnessMode(ctx context.Context, netCfg network.Config, promListen strin pterm.Fatal.Printfln("cannot create ledger dir %s: %v", dbPath, err) } + cstore, err := comments.Open(defaultCommentsRoot()) + if err != nil { + slog.Warn("witness: comments store open failed; CommentFetch disabled", + slog.Any(obs.FieldErr, err)) + } + l, err := ledger.NewWithOptions(dbPath, priv, node.PubSub, ledger.Options{ - Host: node.Host, + Host: node.Host, + Comments: cstore, }) if err != nil { pterm.Fatal.Printfln("ledger open failed: %v", err) @@ -66,11 +73,7 @@ func runWitnessMode(ctx context.Context, netCfg network.Config, promListen strin defer func() { _ = l.Close() }() slog.Info("witness: ledger opened", slog.String("path", dbPath)) - cstore, err := comments.Open(defaultCommentsRoot()) - if err != nil { - slog.Warn("witness: comments store open failed; CommentFetch disabled", - slog.Any(obs.FieldErr, err)) - } else { + if cstore != nil { cserver := comments.NewServer(node.Host, cstore) cserver.Start() defer cserver.Stop() diff --git a/agentfm-go/internal/boss/api.go b/agentfm-go/internal/boss/api.go index fb94c5b..f934d99 100644 --- a/agentfm-go/internal/boss/api.go +++ b/agentfm-go/internal/boss/api.go @@ -40,6 +40,11 @@ type ExecuteRequest struct { // AsyncExecuteRequest is the request body for POST /api/execute/async. // WebhookURL is optional; when empty the Boss finishes the task quietly // and writes artifacts to disk without notifying anyone. +// +// There is no client-supplied task_id: the async endpoint always mints its +// own and returns it in the 202 body ("task_id"). Clients MUST read the +// response for the id — it is the sole handle for artifact retrieval and +// webhook correlation. type AsyncExecuteRequest struct { WorkerID string `json:"worker_id"` Prompt string `json:"prompt"` diff --git a/agentfm-go/internal/boss/api_async.go b/agentfm-go/internal/boss/api_async.go index 31615a4..6c1535d 100644 --- a/agentfm-go/internal/boss/api_async.go +++ b/agentfm-go/internal/boss/api_async.go @@ -87,8 +87,17 @@ func (b *Boss) asyncExecuteHandler(rootCtx context.Context, wg *sync.WaitGroup) return } + if req.WorkerID == "" { + http.Error(w, "worker_id is required", http.StatusBadRequest) + return + } + if req.Prompt == "" { + http.Error(w, "prompt is required", http.StatusBadRequest) + return + } + b.mu.RLock() - _, exists := b.activeWorkers[req.WorkerID] + profile, exists := b.activeWorkers[req.WorkerID] b.mu.RUnlock() if !exists { @@ -102,6 +111,14 @@ func (b *Boss) asyncExecuteHandler(rootCtx context.Context, wg *sync.WaitGroup) return } + // Trust gate BEFORE committing an async slot + goroutine — an + // equivocator or below-floor peer must be refused synchronously + // (same gate as /api/execute and the OpenAI-compat path). + if outcome := b.checkTrust(r.Context(), profile); !outcome.Allowed { + http.Error(w, "dispatch refused: "+outcome.Reason, http.StatusForbidden) + return + } + taskID := newCompletionID("task_") // Acquire an async slot non-blockingly. When MaxInflightAsyncTasks diff --git a/agentfm-go/internal/boss/api_comment_fetch_fallback_test.go b/agentfm-go/internal/boss/api_comment_fetch_fallback_test.go new file mode 100644 index 0000000..408a73b --- /dev/null +++ b/agentfm-go/internal/boss/api_comment_fetch_fallback_test.go @@ -0,0 +1,137 @@ +package boss + +import ( + "encoding/hex" + "fmt" + "net/http/httptest" + "testing" + "time" + + "agentfm/internal/ledger/comments" + "agentfm/internal/network" + "agentfm/internal/types" + "agentfm/test/testutil" +) + +func newFallbackBoss(t *testing.T, opts func(*Boss)) (*Boss, *comments.Store) { + t.Helper() + localStore, err := comments.Open(t.TempDir()) + if err != nil { + t.Fatalf("comments.Open: %v", err) + } + b := &Boss{ + node: &network.MeshNode{Host: testutil.NewHost(t)}, + activeWorkers: make(map[string]types.WorkerProfile), + lastSeen: make(map[string]time.Time), + commentsStore: localStore, + } + if opts != nil { + opts(b) + } + return b, localStore +} + +func TestCommentBodyGet_FetchesFromAuthorOnMiss(t *testing.T) { + authorHost := testutil.NewHost(t) + authorStore, err := comments.Open(t.TempDir()) + if err != nil { + t.Fatalf("comments.Open author: %v", err) + } + srv := comments.NewServer(authorHost, authorStore) + srv.Start() + defer srv.Stop() + + body := []byte("fetched over p2p from the author") + cid, err := authorStore.Put(body) + if err != nil { + t.Fatalf("author Put: %v", err) + } + + subject := testutil.NewHost(t).ID() + readStore := testutil.OpenTestStore(t) + testutil.AppendInboxComment(t, readStore, authorHost, subject, cid) + + b, localStore := newFallbackBoss(t, func(b *Boss) { b.readStore = readStore }) + testutil.ConnectHosts(t, b.node.Host, authorHost) + + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", + fmt.Sprintf("/v1/peers/%s/comments/%s", subject.String(), hex.EncodeToString(cid)), nil) + b.handlePeersForTest(rec, req) + + if rec.Code != 200 { + t.Fatalf("status=%d body=%q", rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); got != string(body) { + t.Fatalf("body mismatch: %q", got) + } + if !localStore.Has(cid) { + t.Fatalf("fetched body was not cached in the local store") + } +} + +func TestCommentBodyGet_FetchesFromRelayWhenAuthorUnknown(t *testing.T) { + relayHost := testutil.NewHost(t) + relayStore, err := comments.Open(t.TempDir()) + if err != nil { + t.Fatalf("comments.Open relay: %v", err) + } + srv := comments.NewServer(relayHost, relayStore) + srv.Start() + defer srv.Stop() + + body := []byte("fetched over p2p from the relay archive") + cid, err := relayStore.Put(body) + if err != nil { + t.Fatalf("relay Put: %v", err) + } + + b, _ := newFallbackBoss(t, func(b *Boss) { b.node.RelayPeerID = relayHost.ID() }) + testutil.ConnectHosts(t, b.node.Host, relayHost) + + subject := testutil.NewHost(t).ID() + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", + fmt.Sprintf("/v1/peers/%s/comments/%s.json", subject.String(), hex.EncodeToString(cid)), nil) + b.handlePeersForTest(rec, req) + + if rec.Code != 200 { + t.Fatalf("status=%d body=%q", rec.Code, rec.Body.String()) + } +} + +func TestCommentBodyGet_400OnStructurallyInvalidCID(t *testing.T) { + relayHost := testutil.NewHost(t) + b, _ := newFallbackBoss(t, func(b *Boss) { b.node.RelayPeerID = relayHost.ID() }) + + subject := testutil.NewHost(t).ID() + for _, cid := range []string{"ab", "12ab34", "ff" + hex.EncodeToString(make([]byte, 33))} { + start := time.Now() + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", + fmt.Sprintf("/v1/peers/%s/comments/%s", subject.String(), cid), nil) + b.handlePeersForTest(rec, req) + + if rec.Code != 400 { + t.Fatalf("cid %q: status=%d; want 400", cid, rec.Code) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("cid %q: malformed CID took %v; must reject without remote fetches", cid, elapsed) + } + } +} + +func TestCommentBodyGet_404WhenNoSourceHasIt(t *testing.T) { + b, _ := newFallbackBoss(t, nil) + + missing := comments.CIDOf([]byte("never stored anywhere")) + subject := testutil.NewHost(t).ID() + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", + fmt.Sprintf("/v1/peers/%s/comments/%s", subject.String(), hex.EncodeToString(missing)), nil) + b.handlePeersForTest(rec, req) + + if rec.Code != 404 { + t.Fatalf("status=%d; want 404", rec.Code) + } +} diff --git a/agentfm-go/internal/boss/api_comments.go b/agentfm-go/internal/boss/api_comments.go index 8bce526..abd6074 100644 --- a/agentfm-go/internal/boss/api_comments.go +++ b/agentfm-go/internal/boss/api_comments.go @@ -1,6 +1,8 @@ package boss import ( + "bytes" + "context" "crypto/sha256" "encoding/base64" "encoding/hex" @@ -16,9 +18,11 @@ import ( "agentfm/internal/ledger/comments" pb "agentfm/internal/ledger/pb" + "agentfm/internal/ledger/store" "agentfm/internal/obs" "github.com/libp2p/go-libp2p/core/peer" + "google.golang.org/protobuf/proto" ) // base64StdDecoder is captured at package scope so the inline @@ -223,6 +227,91 @@ func base64Decode(s string) ([]byte, error) { return base64StdDecoder.DecodeString(s) } +const commentFetchPerPeerTimeout = 10 * time.Second + +// getCommentBody returns the body for cidBytes, first from the local +// store, then — on a miss — fetched over CommentFetchProtocol from the +// comment's author (resolved via the inbox) or the relay, caching any +// fetched body locally. +func (b *Boss) getCommentBody(ctx context.Context, cidBytes []byte) ([]byte, error) { + body, err := b.commentsStore.Get(cidBytes) + if err == nil { + return body, nil + } + if !errors.Is(err, comments.ErrNotFound) && !errors.Is(err, comments.ErrCIDMismatch) { + return nil, err + } + if b.node == nil || b.node.Host == nil { + return nil, err + } + for _, source := range b.commentBodySources(ctx, cidBytes) { + fetchCtx, cancel := context.WithTimeout(ctx, commentFetchPerPeerTimeout) + fetched, ferr := comments.Fetch(fetchCtx, b.node.Host, source, cidBytes) + cancel() + if ferr != nil { + slog.Debug("boss: remote comment body fetch failed", + slog.String("peer", source.String()), + slog.Any(obs.FieldErr, ferr)) + continue + } + if _, perr := b.commentsStore.Put(fetched); perr != nil { + slog.Warn("boss: caching fetched comment body failed", + slog.Any(obs.FieldErr, perr)) + } + return fetched, nil + } + return nil, comments.ErrNotFound +} + +// commentBodySources returns candidate peers to fetch a body from: +// the comment's author first (when resolvable from the inbox), then +// the relay archive. +func (b *Boss) commentBodySources(ctx context.Context, cidBytes []byte) []peer.ID { + self := b.node.Host.ID() + sources := make([]peer.ID, 0, 2) + if b.readStore != nil { + if author, ok := findCommentAuthor(ctx, b.readStore, cidBytes); ok && author != self { + sources = append(sources, author) + } + } + if relay := b.node.RelayPeerID; relay != "" && relay != self { + if len(sources) == 0 || sources[0] != relay { + sources = append(sources, relay) + } + } + return sources +} + +// errStopAuthorScan is the sentinel findCommentAuthor's callback returns +// to stop iteration as soon as the matching Comment is found. +var errStopAuthorScan = errors.New("stop author scan") + +// findCommentAuthor scans the inbox for a Comment entry whose text_cid +// matches cidBytes and returns its rater peer ID. +func findCommentAuthor(ctx context.Context, s *store.Store, cidBytes []byte) (peer.ID, bool) { + var author peer.ID + found := false + err := s.IterateAllInboxEntries(ctx, func(e *store.InboxEntry) error { + var signed pb.SignedEntry + if uerr := proto.Unmarshal(e.Payload, &signed); uerr != nil { + return nil + } + if body, ok := signed.GetBody().(*pb.SignedEntry_Comment); ok && + body.Comment != nil && bytes.Equal(body.Comment.TextCid, cidBytes) { + author = peer.ID(body.Comment.RaterPeerId) + found = true + return errStopAuthorScan + } + return nil + }) + if err != nil && !errors.Is(err, errStopAuthorScan) { + slog.Warn("boss: comment author inbox scan failed", + slog.Any(obs.FieldErr, err)) + return "", false + } + return author, found +} + // handleCommentBodyGet services GET /v1/peers/{id}/comments/{cid}. // // The CID is hex-encoded (same format as CommentSubmitResponse.CID and the @@ -231,7 +320,7 @@ func base64Decode(s string) ([]byte, error) { // // Errors: // - 400: CID is malformed hex -// - 404: CID not found in local body store +// - 404: CID not found locally, from the author, or from the relay // - 503: commentsStore not wired on this boss func (b *Boss) handleCommentBodyGet(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { @@ -249,13 +338,13 @@ func (b *Boss) handleCommentBodyGet(w http.ResponseWriter, r *http.Request) { writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_request", "missing CID in path") return } - cidBytes, err := hex.DecodeString(cidHex) + cidBytes, err := comments.ParseCIDString(cidHex) if err != nil { - writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_cid", "CID is not valid hex: "+err.Error()) + writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_cid", "CID is not a valid multihash: "+err.Error()) return } - body, err := b.commentsStore.Get(cidBytes) + body, err := b.getCommentBody(r.Context(), cidBytes) if err != nil { if errors.Is(err, comments.ErrNotFound) || errors.Is(err, comments.ErrCIDMismatch) { writeOpenAIError(w, http.StatusNotFound, errTypeInvalidRequest, "comment_not_found", "comment body not found for this CID") @@ -317,13 +406,13 @@ func (b *Boss) handleCommentBodyGetJSON(w http.ResponseWriter, r *http.Request) writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_request", "missing CID in path") return } - cidBytes, err := hex.DecodeString(cidHex) + cidBytes, err := comments.ParseCIDString(cidHex) if err != nil { - writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_cid", "CID is not valid hex: "+err.Error()) + writeOpenAIError(w, http.StatusBadRequest, errTypeInvalidRequest, "bad_cid", "CID is not a valid multihash: "+err.Error()) return } - body, err := b.commentsStore.Get(cidBytes) + body, err := b.getCommentBody(r.Context(), cidBytes) if err != nil { if errors.Is(err, comments.ErrNotFound) || errors.Is(err, comments.ErrCIDMismatch) { writeOpenAIError(w, http.StatusNotFound, errTypeInvalidRequest, "comment_not_found", "comment body not found for this CID") diff --git a/agentfm-go/internal/boss/api_dispatch_trust_test.go b/agentfm-go/internal/boss/api_dispatch_trust_test.go new file mode 100644 index 0000000..49b1fc2 --- /dev/null +++ b/agentfm-go/internal/boss/api_dispatch_trust_test.go @@ -0,0 +1,107 @@ +package boss + +import ( + "net/http/httptest" + "strings" + "testing" + + "agentfm/internal/network" + "agentfm/internal/types" + "agentfm/test/testutil" +) + +// seededDispatchBoss returns a Boss with one seeded online worker whose +// peer ID is a real libp2p ID (so peer.Decode succeeds and the trust gate +// runs). The returned peer-id string is the worker_id to dispatch to. +func seededDispatchBoss(t *testing.T) (*Boss, string) { + t.Helper() + h := testutil.NewHost(t) + b := NewForTest(&network.MeshNode{Host: h}) + workerID := testutil.NewHost(t).ID().String() + b.SeedWorker(types.WorkerProfile{PeerID: workerID, AgentName: "Victim", Status: "AVAILABLE"}) + return b, workerID +} + +func dispatchBody(workerID, prompt string) string { + return `{"worker_id":"` + workerID + `","prompt":"` + prompt + `","task_id":"task_trust_gate"}` +} + +func TestExecute_TrustGateBlocksEquivocator(t *testing.T) { + b, workerID := seededDispatchBoss(t) + b.ledger = alwaysEquivocatorLedger{} + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute", strings.NewReader(dispatchBody(workerID, "hi"))) + b.ServeHTTPExecute(rec, req) + + if rec.Code != 403 { + t.Fatalf("equivocator dispatch: status=%d, want 403; body=%q", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "equivocator") { + t.Errorf("body should name the refusal reason; got %q", rec.Body.String()) + } +} + +func TestExecute_TrustGateBlocksBelowFloor(t *testing.T) { + b, workerID := seededDispatchBoss(t) + b.reputationFloor = -0.5 + b.SetReputationScoreForTest(workerID, -0.9) + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute", strings.NewReader(dispatchBody(workerID, "hi"))) + b.ServeHTTPExecute(rec, req) + + if rec.Code != 403 { + t.Fatalf("below-floor dispatch: status=%d, want 403; body=%q", rec.Code, rec.Body.String()) + } +} + +func TestExecuteAsync_TrustGateBlocksEquivocator(t *testing.T) { + b, workerID := seededDispatchBoss(t) + b.ledger = alwaysEquivocatorLedger{} + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute/async", strings.NewReader(dispatchBody(workerID, "hi"))) + b.ServeHTTPExecuteAsync(rec, req) + + if rec.Code != 403 { + t.Fatalf("async equivocator dispatch: status=%d, want 403; body=%q", rec.Code, rec.Body.String()) + } +} + +func TestExecute_MissingWorkerIDIs400(t *testing.T) { + b, _ := seededDispatchBoss(t) + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute", strings.NewReader(`{}`)) + b.ServeHTTPExecute(rec, req) + + if rec.Code != 400 { + t.Fatalf("empty body: status=%d, want 400; body=%q", rec.Code, rec.Body.String()) + } +} + +func TestExecute_MissingPromptIs400(t *testing.T) { + b, workerID := seededDispatchBoss(t) + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute", + strings.NewReader(`{"worker_id":"`+workerID+`"}`)) + b.ServeHTTPExecute(rec, req) + + if rec.Code != 400 { + t.Fatalf("missing prompt: status=%d, want 400; body=%q", rec.Code, rec.Body.String()) + } +} + +func TestExecuteAsync_MissingWorkerIDIs400(t *testing.T) { + b, _ := seededDispatchBoss(t) + + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/execute/async", strings.NewReader(`{}`)) + b.ServeHTTPExecuteAsync(rec, req) + + if rec.Code != 400 { + t.Fatalf("async empty body: status=%d, want 400; body=%q", rec.Code, rec.Body.String()) + } +} diff --git a/agentfm-go/internal/boss/api_handlers.go b/agentfm-go/internal/boss/api_handlers.go index c9ecad4..df18d89 100644 --- a/agentfm-go/internal/boss/api_handlers.go +++ b/agentfm-go/internal/boss/api_handlers.go @@ -185,6 +185,15 @@ func (b *Boss) handleExecuteTask(w http.ResponseWriter, r *http.Request) { return } + if req.WorkerID == "" { + http.Error(w, "worker_id is required", http.StatusBadRequest) + return + } + if req.Prompt == "" { + http.Error(w, "prompt is required", http.StatusBadRequest) + return + } + if req.TaskID == "" { req.TaskID = newCompletionID("task_") } @@ -194,7 +203,7 @@ func (b *Boss) handleExecuteTask(w http.ResponseWriter, r *http.Request) { } b.mu.RLock() - _, exists := b.activeWorkers[req.WorkerID] + profile, exists := b.activeWorkers[req.WorkerID] b.mu.RUnlock() if !exists { @@ -209,6 +218,15 @@ func (b *Boss) handleExecuteTask(w http.ResponseWriter, r *http.Request) { return } + // Trust gate: equivocators and below-floor peers are refused before any + // dial. The OpenAI-compat path already gates here; direct /api/execute + // (and the desktop Dispatch button, which uses it) must not be a bypass. + if outcome := b.checkTrust(r.Context(), profile); !outcome.Allowed { + status = metrics.StatusRejected + http.Error(w, "dispatch refused: "+outcome.Reason, http.StatusForbidden) + return + } + pterm.Info.Printfln("📡 API Gateway routing task %s to Worker %s...", shortID(req.TaskID, 8), pterm.Cyan(peerID.String()[:8])) diff --git a/agentfm-go/internal/boss/comment_body_test.go b/agentfm-go/internal/boss/comment_body_test.go index e960b3a..1fe4f37 100644 --- a/agentfm-go/internal/boss/comment_body_test.go +++ b/agentfm-go/internal/boss/comment_body_test.go @@ -163,8 +163,8 @@ func TestCommentBodyHydration_BadHexCID(t *testing.T) { // TestCommentBodyHydration_NotFound returns 404. func TestCommentBodyHydration_NotFound(t *testing.T) { cbr := newCommentBodyRig(t) - // CID that doesn't exist: 34 zero bytes as hex. - cidHex := hex.EncodeToString(make([]byte, 34)) + // Structurally valid CID whose body was never stored. + cidHex := comments.CIDString(comments.CIDOf([]byte("no body stored for this"))) rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/v1/peers/"+cbr.subject.String()+"/comments/"+cidHex, nil) diff --git a/agentfm-go/internal/boss/testing.go b/agentfm-go/internal/boss/testing.go index 8e0fb82..ae79f6d 100644 --- a/agentfm-go/internal/boss/testing.go +++ b/agentfm-go/internal/boss/testing.go @@ -39,6 +39,14 @@ func (b *Boss) ServeHTTPExecute(w http.ResponseWriter, r *http.Request) { b.handleExecuteTask(w, r) } +// ServeHTTPExecuteAsync exposes the async execute handler to test packages. +// Reject paths (validation, trust gate) return before any goroutine spawns, +// so no WaitGroup drain is required. +func (b *Boss) ServeHTTPExecuteAsync(w http.ResponseWriter, r *http.Request) { + var wg sync.WaitGroup + b.asyncExecuteHandler(context.Background(), &wg)(w, r) +} + // HostForTest returns the boss's underlying libp2p host identity so test // helpers can write ledger entries attributed to the boss's own peer ID. // Must only be used in test code. diff --git a/agentfm-go/internal/ledger/bodyfetch.go b/agentfm-go/internal/ledger/bodyfetch.go new file mode 100644 index 0000000..52d8fae --- /dev/null +++ b/agentfm-go/internal/ledger/bodyfetch.go @@ -0,0 +1,165 @@ +package ledger + +import ( + "context" + "errors" + "log/slog" + "time" + + "agentfm/internal/ledger/comments" + pb "agentfm/internal/ledger/pb" + "agentfm/internal/ledger/store" + "agentfm/internal/obs" + + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/peer" + + "google.golang.org/protobuf/proto" +) + +const ( + bodyFetchQueueCap = 64 + bodyFetchTimeout = 30 * time.Second + + bodyBackfillInitialDelay = time.Minute + bodyBackfillInterval = 10 * time.Minute + bodyBackfillMaxPerSweep = 256 +) + +// errStopBackfillScan is the sentinel the backfill scan callback returns +// to stop iterating once the per-sweep cap is reached. +var errStopBackfillScan = errors.New("stop backfill scan") + +type bodyFetchJob struct { + author peer.ID + cid []byte +} + +type bodyFetcher struct { + host host.Host + store *comments.Store + jobs chan bodyFetchJob + done chan struct{} + backfillDone chan struct{} +} + +func newBodyFetcher(h host.Host, s *comments.Store) *bodyFetcher { + return &bodyFetcher{ + host: h, + store: s, + jobs: make(chan bodyFetchJob, bodyFetchQueueCap), + done: make(chan struct{}), + backfillDone: make(chan struct{}), + } +} + +func (f *bodyFetcher) enqueue(entry *pb.SignedEntry) { + c := commentOf(entry) + if c == nil || len(c.TextCid) == 0 || f.store.Has(c.TextCid) { + return + } + select { + case f.jobs <- bodyFetchJob{author: peer.ID(c.RaterPeerId), cid: c.TextCid}: + default: + slog.Debug("ledger: comment body fetch queue full; deferring to backfill sweep", + slog.String("author", peer.ID(c.RaterPeerId).String())) + } +} + +func (f *bodyFetcher) run(ctx context.Context) { + defer close(f.done) + for { + select { + case <-ctx.Done(): + return + case job := <-f.jobs: + f.fetchOne(ctx, job) + } + } +} + +// runBackfill periodically re-scans the inbox for Comment entries whose +// bodies are still missing from the store — bodies dropped by a full +// live queue, or whose author was unreachable at gossip time — and +// re-fetches them. The first sweep runs shortly after startup so a +// restarted node heals gaps without waiting a full interval. +func (f *bodyFetcher) runBackfill(ctx context.Context, s *store.Store) { + defer close(f.backfillDone) + timer := time.NewTimer(bodyBackfillInitialDelay) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + } + f.backfillOnce(ctx, s) + timer.Reset(bodyBackfillInterval) + } +} + +// backfillOnce collects up to bodyBackfillMaxPerSweep missing (author, +// cid) pairs from the inbox, then fetches them serially. Collect and +// fetch are separate phases so no network I/O happens inside the SQLite +// row iteration. +func (f *bodyFetcher) backfillOnce(ctx context.Context, s *store.Store) { + missing := make([]bodyFetchJob, 0, 16) + truncated := false + err := s.IterateAllInboxEntries(ctx, func(e *store.InboxEntry) error { + var signed pb.SignedEntry + if uerr := proto.Unmarshal(e.Payload, &signed); uerr != nil { + return nil + } + c := commentOf(&signed) + if c == nil || len(c.TextCid) == 0 || f.store.Has(c.TextCid) { + return nil + } + if len(missing) >= bodyBackfillMaxPerSweep { + truncated = true + return errStopBackfillScan + } + missing = append(missing, bodyFetchJob{author: peer.ID(c.RaterPeerId), cid: c.TextCid}) + return nil + }) + if err != nil && !errors.Is(err, errStopBackfillScan) { + slog.Debug("ledger: comment body backfill scan failed", + slog.Any(obs.FieldErr, err)) + return + } + if truncated { + slog.Debug("ledger: comment body backfill hit per-sweep cap; remainder picked up next sweep", + slog.Int("cap", bodyBackfillMaxPerSweep)) + } + for _, job := range missing { + if ctx.Err() != nil { + return + } + f.fetchOne(ctx, job) + } +} + +func (f *bodyFetcher) fetchOne(ctx context.Context, job bodyFetchJob) { + if f.store.Has(job.cid) { + return + } + fetchCtx, cancel := context.WithTimeout(ctx, bodyFetchTimeout) + body, err := comments.Fetch(fetchCtx, f.host, job.author, job.cid) + cancel() + if err != nil { + slog.Debug("ledger: comment body fetch failed", + slog.String("author", job.author.String()), + slog.Any(obs.FieldErr, err)) + return + } + if _, err := f.store.Put(body); err != nil { + slog.Warn("ledger: comment body persist failed", + slog.Any(obs.FieldErr, err)) + } +} + +func commentOf(entry *pb.SignedEntry) *pb.Comment { + if body, ok := entry.GetBody().(*pb.SignedEntry_Comment); ok { + return body.Comment + } + return nil +} diff --git a/agentfm-go/internal/ledger/bodyfetch_test.go b/agentfm-go/internal/ledger/bodyfetch_test.go new file mode 100644 index 0000000..a7a96e0 --- /dev/null +++ b/agentfm-go/internal/ledger/bodyfetch_test.go @@ -0,0 +1,76 @@ +package ledger + +import ( + "context" + "path/filepath" + "testing" + + "agentfm/internal/ledger/comments" + "agentfm/test/testutil" +) + +func TestBodyBackfill_FetchesMissingBodies(t *testing.T) { + hosts := testutil.NewConnectedMesh(t, 3) + authorHost, archiveHost, subjectHost := hosts[0], hosts[1], hosts[2] + + authorStore, err := comments.Open(filepath.Join(t.TempDir(), "author")) + if err != nil { + t.Fatalf("open author store: %v", err) + } + srv := comments.NewServer(authorHost, authorStore) + srv.Start() + t.Cleanup(srv.Stop) + + body := []byte("backfilled after the live fetch was missed") + cid, err := authorStore.Put(body) + if err != nil { + t.Fatalf("author Put: %v", err) + } + + s := testutil.OpenTestStore(t) + testutil.AppendInboxComment(t, s, authorHost, subjectHost.ID(), cid) + + archiveStore, err := comments.Open(filepath.Join(t.TempDir(), "archive")) + if err != nil { + t.Fatalf("open archive store: %v", err) + } + f := newBodyFetcher(archiveHost, archiveStore) + f.backfillOnce(context.Background(), s) + + if !archiveStore.Has(cid) { + t.Fatal("backfillOnce should have fetched and stored the missing body") + } + + got, err := archiveStore.Get(cid) + if err != nil { + t.Fatalf("Get after backfill: %v", err) + } + if string(got) != string(body) { + t.Fatalf("backfilled body mismatch: %q", got) + } +} + +func TestBodyBackfill_SkipsBodiesAlreadyStored(t *testing.T) { + hosts := testutil.NewConnectedMesh(t, 2) + authorHost, archiveHost := hosts[0], hosts[1] + + s := testutil.OpenTestStore(t) + archiveStore, err := comments.Open(filepath.Join(t.TempDir(), "archive")) + if err != nil { + t.Fatalf("open archive store: %v", err) + } + + body := []byte("already present locally") + cid, err := archiveStore.Put(body) + if err != nil { + t.Fatalf("archive Put: %v", err) + } + testutil.AppendInboxComment(t, s, authorHost, archiveHost.ID(), cid) + + f := newBodyFetcher(archiveHost, archiveStore) + f.backfillOnce(context.Background(), s) + + if !archiveStore.Has(cid) { + t.Fatal("existing body must remain in the store") + } +} diff --git a/agentfm-go/internal/ledger/comments/comments_test.go b/agentfm-go/internal/ledger/comments/comments_test.go index f44717f..8830ff3 100644 --- a/agentfm-go/internal/ledger/comments/comments_test.go +++ b/agentfm-go/internal/ledger/comments/comments_test.go @@ -8,9 +8,13 @@ import ( "path/filepath" "strings" "testing" + "time" "agentfm/internal/ledger/comments" + "agentfm/internal/network" "agentfm/test/testutil" + + libnet "github.com/libp2p/go-libp2p/core/network" ) func TestCIDOf_Deterministic(t *testing.T) { @@ -166,6 +170,63 @@ func TestFetch_RoundTrip(t *testing.T) { } } +func TestFetch_CtxBoundsStalledServer(t *testing.T) { + hosts := testutil.NewConnectedMesh(t, 2) + srvHost, cliHost := hosts[0], hosts[1] + + stall := make(chan struct{}) + t.Cleanup(func() { close(stall) }) + srvHost.SetStreamHandler(network.CommentFetchProtocol, func(s libnet.Stream) { + <-stall + _ = s.Reset() + }) + + cid := comments.CIDOf([]byte("body the server will never send")) + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + + start := time.Now() + _, err := comments.Fetch(ctx, cliHost, srvHost.ID(), cid) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("Fetch against a stalled server should fail") + } + if elapsed > 5*time.Second { + t.Fatalf("Fetch ignored ctx deadline: took %v against a stalled server (want ~1s)", elapsed) + } +} + +func TestFetch_CancelUnblocksImmediately(t *testing.T) { + hosts := testutil.NewConnectedMesh(t, 2) + srvHost, cliHost := hosts[0], hosts[1] + + stall := make(chan struct{}) + t.Cleanup(func() { close(stall) }) + srvHost.SetStreamHandler(network.CommentFetchProtocol, func(s libnet.Stream) { + <-stall + _ = s.Reset() + }) + + cid := comments.CIDOf([]byte("never delivered")) + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(300 * time.Millisecond) + cancel() + }() + + start := time.Now() + _, err := comments.Fetch(ctx, cliHost, srvHost.ID(), cid) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("Fetch should fail when ctx is cancelled mid-flight") + } + if elapsed > 5*time.Second { + t.Fatalf("Fetch did not unblock on ctx cancel: took %v (want ~300ms)", elapsed) + } +} + func TestFetch_NotFound(t *testing.T) { hosts := testutil.NewConnectedMesh(t, 2) srvHost, cliHost := hosts[0], hosts[1] diff --git a/agentfm-go/internal/ledger/comments/fetch.go b/agentfm-go/internal/ledger/comments/fetch.go index e282304..8494660 100644 --- a/agentfm-go/internal/ledger/comments/fetch.go +++ b/agentfm-go/internal/ledger/comments/fetch.go @@ -116,6 +116,10 @@ func writeNotFound(w io.Writer, reason string) { // body for cid. Returns (body, nil) on success; ErrNotFound when // the remote doesn't have the body; or a transport / decode error. // +// The stream deadline is the sooner of fetchTimeout and ctx's own +// deadline, and ctx cancellation resets the stream immediately — a +// stalled remote can never pin the caller past its context bound. +// // Validates the returned body against cid before returning — // callers can trust the bytes match what they asked for without // re-hashing. @@ -125,7 +129,13 @@ func Fetch(ctx context.Context, h host.Host, remote peer.ID, cid []byte) ([]byte return nil, fmt.Errorf("comments fetch: open stream: %w", err) } defer func() { _ = s.Close() }() - if err := s.SetDeadline(time.Now().Add(fetchTimeout)); err != nil { + stop := context.AfterFunc(ctx, func() { _ = s.Reset() }) + defer stop() + deadline := time.Now().Add(fetchTimeout) + if d, ok := ctx.Deadline(); ok && d.Before(deadline) { + deadline = d + } + if err := s.SetDeadline(deadline); err != nil { return nil, fmt.Errorf("comments fetch: set deadline: %w", err) } diff --git a/agentfm-go/internal/ledger/impl.go b/agentfm-go/internal/ledger/impl.go index ecd9672..1f9a10a 100644 --- a/agentfm-go/internal/ledger/impl.go +++ b/agentfm-go/internal/ledger/impl.go @@ -60,6 +60,12 @@ type ledgerImpl struct { // implementation has no init that can fail at this point). inbox *inbox.Inbox + // bodyFetcher replicates comment bodies referenced by gossip-accepted + // Comment entries. nil unless Options.Comments + Host + pubsub were + // all provided. Set once in newImpl before the subscriber goroutine + // starts; never mutated afterwards. + bodyFetcher *bodyFetcher + // Subscriber goroutine lifecycle. Only populated when ps != nil. subCtx context.Context subCancel context.CancelFunc @@ -136,6 +142,11 @@ func newImpl(path string, key crypto.PrivKey, ps *pubsub.PubSub, opts Options) ( l.sub = sub l.subCtx, l.subCancel = context.WithCancel(context.Background()) l.subDone = make(chan struct{}) + if opts.Host != nil && opts.Comments != nil { + l.bodyFetcher = newBodyFetcher(opts.Host, opts.Comments) + go l.bodyFetcher.run(l.subCtx) + go l.bodyFetcher.runBackfill(l.subCtx, l.store) + } // Pass sub + subCtx + subDone as args so the goroutine holds // stable references — Close() races to nil the struct fields // during shutdown, and we don't want the goroutine reading @@ -665,6 +676,10 @@ func (l *ledgerImpl) runSubscriber(ctx context.Context, sub *pubsub.Subscription // is appropriate so operators can correlate when needed. slog.Debug("ledger: gossip entry rejected by inbox", slog.Any(obs.FieldErr, err)) + continue + } + if l.bodyFetcher != nil { + l.bodyFetcher.enqueue(&entry) } } } @@ -711,6 +726,10 @@ func (l *ledgerImpl) Close() error { if subDone != nil { <-subDone } + if l.bodyFetcher != nil { + <-l.bodyFetcher.done + <-l.bodyFetcher.backfillDone + } if equivSub != nil { equivSub.Cancel() diff --git a/agentfm-go/internal/ledger/ledger.go b/agentfm-go/internal/ledger/ledger.go index 95ead2a..d4d0faa 100644 --- a/agentfm-go/internal/ledger/ledger.go +++ b/agentfm-go/internal/ledger/ledger.go @@ -3,6 +3,7 @@ package ledger import ( "context" + "agentfm/internal/ledger/comments" pb "agentfm/internal/ledger/pb" "agentfm/internal/ledger/store" @@ -17,6 +18,12 @@ type Options struct { // and HeadFetchProtocol stream handlers (P2-5, P5-1). nil disables // both handlers (local-only / test mode). Host host.Host + + // Comments, when non-nil (and Host + pubsub are also set), makes + // this ledger fetch and persist the body of every Comment entry + // accepted from gossip, pulling it from the entry's author via + // CommentFetchProtocol. nil disables body replication. + Comments *comments.Store } // IsHeadValid reports whether head carries at least threshold diff --git a/agentfm-go/internal/network/discovery.go b/agentfm-go/internal/network/discovery.go index b2f250c..48d9df4 100644 --- a/agentfm-go/internal/network/discovery.go +++ b/agentfm-go/internal/network/discovery.go @@ -18,6 +18,16 @@ import ( "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/client" ) +// lighthouseConnTag protects the relay connection from connection-manager +// trimming: without it, the connmgr prunes the (usually idle) lighthouse +// connection under pressure and the node silently loses its relay. +const lighthouseConnTag = "lighthouse" + +// LighthouseReconnectInterval is how often maintainLighthouseConnection +// re-checks and, if dropped, re-dials the relay. Lives here (not in the +// gitignored constants.go) so it stays version-controlled. +const LighthouseReconnectInterval = 30 * time.Second + // connectToLighthouse dials the bootstrap relay and, on success, reserves // a circuit-relay slot on it so this node can be reached via p2p-circuit // when direct NAT traversal fails. Both steps are bounded by @@ -34,6 +44,9 @@ func connectToLighthouse(ctx context.Context, h host.Host, relayInfo *peer.AddrI ) return } + // Protect the relay from connmgr trimming so an idle lighthouse + // connection survives connection-count pressure. Idempotent. + h.ConnManager().Protect(relayInfo.ID, lighthouseConnTag) fmt.Println("✅ Successfully connected to Bootstrap Node!") reserveCtx, reserveCancel := context.WithTimeout(ctx, StreamDialTimeout) @@ -48,6 +61,44 @@ func connectToLighthouse(ctx context.Context, h host.Host, relayInfo *peer.AddrI } } +// maintainLighthouseConnection re-dials (and re-reserves) the relay whenever +// the direct connection drops — an idle prune, a reservation TTL expiry, or a +// transient network blip. Without it the node loses its relay for good after +// the first disconnect, and boss /v1/about reports the relay as unreachable +// ("Connecting to relay…") permanently. Exits on ctx cancellation. +func maintainLighthouseConnection(ctx context.Context, h host.Host, relayInfo *peer.AddrInfo, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if h.Network().Connectedness(relayInfo.ID) == netcore.Connected { + continue + } + slog.Warn("lighthouse connection lost; re-dialing", + slog.String(obs.FieldPeerID, relayInfo.ID.String())) + dialCtx, dialCancel := context.WithTimeout(ctx, StreamDialTimeout) + err := h.Connect(dialCtx, *relayInfo) + dialCancel() + if err != nil { + slog.Warn("lighthouse re-dial failed", + slog.Any(obs.FieldErr, err), + slog.String(obs.FieldPeerID, relayInfo.ID.String())) + continue + } + h.ConnManager().Protect(relayInfo.ID, lighthouseConnTag) + reserveCtx, reserveCancel := context.WithTimeout(ctx, StreamDialTimeout) + if _, err := client.Reserve(reserveCtx, h, *relayInfo); err != nil { + slog.Warn("lighthouse re-reservation failed", + slog.Any(obs.FieldErr, err)) + } + reserveCancel() + } + } +} + // startDiscovery wires up both discovery mechanisms: mDNS for same-LAN // peers and DHT rendezvous for the wider internet. Workers advertise // themselves under the rendezvous string; Boss nodes actively search for diff --git a/agentfm-go/internal/network/discovery_test.go b/agentfm-go/internal/network/discovery_test.go new file mode 100644 index 0000000..05aeead --- /dev/null +++ b/agentfm-go/internal/network/discovery_test.go @@ -0,0 +1,65 @@ +package network + +import ( + "context" + "testing" + "time" + + "agentfm/test/testutil" + + netcore "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" +) + +// TestMaintainLighthouse_ReconnectsAfterDrop proves the relay-connection +// keepalive re-dials after the direct connection drops (idle prune / blip). +// Without the maintenance loop the node loses its relay for good, which is +// what left the desktop toolbar stuck on "Connecting to relay…". +func TestMaintainLighthouse_ReconnectsAfterDrop(t *testing.T) { + a := testutil.NewHost(t) + relay := testutil.NewHost(t) + testutil.ConnectHosts(t, a, relay) + + relayInfo := &peer.AddrInfo{ID: relay.ID(), Addrs: relay.Addrs()} + if a.Network().Connectedness(relay.ID()) != netcore.Connected { + t.Fatal("precondition: A should be connected to the relay") + } + + // Drop the connection the way an idle prune or network blip would. + if err := a.Network().ClosePeer(relay.ID()); err != nil { + t.Fatalf("close peer: %v", err) + } + testutil.Eventually(t, 3*time.Second, func() bool { + return a.Network().Connectedness(relay.ID()) != netcore.Connected + }, "A should be disconnected after ClosePeer") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go maintainLighthouseConnection(ctx, a, relayInfo, 200*time.Millisecond) + + testutil.Eventually(t, 8*time.Second, func() bool { + return a.Network().Connectedness(relay.ID()) == netcore.Connected + }, "maintenance loop should re-dial the dropped relay connection") +} + +// TestMaintainLighthouse_ExitsOnContextCancel guards the goroutine-lifecycle +// rule: the loop must return promptly when its context is cancelled. +func TestMaintainLighthouse_ExitsOnContextCancel(t *testing.T) { + a := testutil.NewHost(t) + relay := testutil.NewHost(t) + relayInfo := &peer.AddrInfo{ID: relay.ID(), Addrs: relay.Addrs()} + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + maintainLighthouseConnection(ctx, a, relayInfo, 200*time.Millisecond) + close(done) + }() + + cancel() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("maintenance loop did not exit within 3s of context cancel") + } +} diff --git a/agentfm-go/internal/network/p2p.go b/agentfm-go/internal/network/p2p.go index bd9115a..abc38dc 100644 --- a/agentfm-go/internal/network/p2p.go +++ b/agentfm-go/internal/network/p2p.go @@ -70,6 +70,9 @@ func Setup(ctx context.Context, cfg Config) (*MeshNode, error) { } else { relayPeerID = relayInfo.ID connectToLighthouse(ctx, h, relayInfo) + // Keep the relay connection alive for the node's lifetime: + // re-dial on drop so the mesh doesn't silently lose its relay. + go maintainLighthouseConnection(ctx, h, relayInfo, LighthouseReconnectInterval) } } diff --git a/agentfm-go/test/integration/relay_comment_body_test.go b/agentfm-go/test/integration/relay_comment_body_test.go new file mode 100644 index 0000000..87048c6 --- /dev/null +++ b/agentfm-go/test/integration/relay_comment_body_test.go @@ -0,0 +1,98 @@ +package integration + +import ( + "bytes" + "context" + "path/filepath" + "testing" + "time" + + "agentfm/internal/ledger" + "agentfm/internal/ledger/comments" + pb "agentfm/internal/ledger/pb" + "agentfm/test/testutil" + + "github.com/libp2p/go-libp2p/core/peer" +) + +func TestRelay_StoresAndServesCommentBodies(t *testing.T) { + tmp := t.TempDir() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + relayKey := mintEdKey(t) + bossKey := mintEdKey(t) + + relayHost := testutil.NewHostWithKey(t, relayKey) + bossHost := testutil.NewHostWithKey(t, bossKey) + testutil.ConnectHosts(t, relayHost, bossHost) + + bossPeerID, err := peer.IDFromPrivateKey(bossKey) + if err != nil { + t.Fatalf("boss peer id: %v", err) + } + + relayPS, bossPS := newPubSubPair(t, ctx, relayHost, bossHost) + + bossStore, err := comments.Open(filepath.Join(tmp, "boss_comments")) + if err != nil { + t.Fatalf("open boss comments store: %v", err) + } + bossSrv := comments.NewServer(bossHost, bossStore) + bossSrv.Start() + defer bossSrv.Stop() + + body := []byte("solid worker, fast artifact turnaround") + cid, err := bossStore.Put(body) + if err != nil { + t.Fatalf("put body on boss: %v", err) + } + + relayStore, err := comments.Open(filepath.Join(tmp, "relay_comments")) + if err != nil { + t.Fatalf("open relay comments store: %v", err) + } + relaySrv := comments.NewServer(relayHost, relayStore) + relaySrv.Start() + defer relaySrv.Stop() + + arch, err := ledger.NewWithOptions(filepath.Join(tmp, "relay_ledger.db"), relayKey, relayPS, + ledger.Options{Host: relayHost, Comments: relayStore}) + if err != nil { + t.Fatalf("open archive ledger: %v", err) + } + defer func() { _ = arch.Close() }() + + time.Sleep(800 * time.Millisecond) + + g, err := ledger.NewWithOptions(filepath.Join(tmp, "boss_ledger.db"), bossKey, bossPS, + ledger.Options{Host: bossHost}) + if err != nil { + t.Fatalf("open boss ledger: %v", err) + } + defer func() { _ = g.Close() }() + + comment := &pb.SignedEntry{Body: &pb.SignedEntry_Comment{Comment: &pb.Comment{ + RaterPeerId: []byte(bossPeerID), + SubjectPeerId: bytes.Repeat([]byte{0xcd}, 32), + Language: "en", + TextCid: cid, + TimestampUnixNs: time.Now().UnixNano(), + }}} + if _, err := g.Append(ctx, comment); err != nil { + t.Fatalf("boss Append comment: %v", err) + } + + testutil.Eventually(t, 10*time.Second, func() bool { + return relayStore.Has(cid) + }, "relay should fetch and store the comment body after ingesting the envelope") + + got, err := comments.Fetch(ctx, bossHost, relayHost.ID(), cid) + if err != nil { + t.Fatalf("fetch body from relay: %v", err) + } + if !bytes.Equal(got, body) { + t.Fatalf("relay served wrong body: got %q want %q", got, body) + } +} diff --git a/agentfm-go/test/testutil/ledger.go b/agentfm-go/test/testutil/ledger.go index 9592111..e601c8d 100644 --- a/agentfm-go/test/testutil/ledger.go +++ b/agentfm-go/test/testutil/ledger.go @@ -84,4 +84,27 @@ func AppendInboxRating(t testing.TB, s *store.Store, rater host.Host, subject pe } } +// AppendInboxComment inserts a Comment entry into the store's INBOX log, +// simulating a comment envelope received from another peer over gossip. +func AppendInboxComment(t testing.TB, s *store.Store, rater host.Host, subject peer.ID, textCID []byte) { + t.Helper() + entry := &pb.SignedEntry{Body: &pb.SignedEntry_Comment{Comment: &pb.Comment{ + RaterPeerId: []byte(rater.ID()), + SubjectPeerId: []byte(subject), + Language: "en", + TextCid: textCID, + TimestampUnixNs: time.Now().UnixNano(), + PrevHash: make([]byte, 32), + }}} + payload, err := proto.Marshal(entry) + if err != nil { + t.Fatalf("testutil.AppendInboxComment marshal: %v", err) + } + var hash [32]byte + copy(hash[:], payload) + if err := s.InsertInboxEntry(ctx2(), []byte(rater.ID()), hash, [32]byte{}, payload); err != nil { + t.Fatalf("testutil.AppendInboxComment InsertInboxEntry: %v", err) + } +} + func ctx2() context.Context { return context.Background() } diff --git a/agentfm-python/src/agentfm/peers.py b/agentfm-python/src/agentfm/peers.py index 0942a3a..be4692f 100644 --- a/agentfm-python/src/agentfm/peers.py +++ b/agentfm-python/src/agentfm/peers.py @@ -2,16 +2,17 @@ Surface: - client.peers.list(include_offline=False) -> list[KnownPeer] - client.peers.get(peer_id) -> PeerSummary - client.peers.log(peer_id, limit=50, offset=0) -> list[PeerEntry] - client.peers.comment_body(peer_id, cid) -> str + client.peers.list(include_offline=False) -> list[KnownPeer] + client.peers.get(peer_id) -> PeerSummary + client.peers.log(peer_id, limit=50, offset=0) -> list[PeerEntry] + client.peers.comment_body(peer_id, cid) -> str + client.peers.comment(peer_id, text, rating=None) -> CommentSubmitResult """ from __future__ import annotations from dataclasses import dataclass -from typing import List, Optional +from typing import Any, Dict, List, Optional import httpx @@ -52,6 +53,21 @@ class PeerSummary: rater_summary: Optional[RaterSummary] = None +@dataclass(frozen=True) +class CommentSubmitResult: + """Returned by ``peers.comment`` after a self-signed submission.""" + + cid: str + ledger_hash: str + + +def _comment_payload(text: str, language: str, rating: Optional[float]) -> Dict[str, Any]: + payload: Dict[str, Any] = {"text": text, "language": language} + if rating is not None: + payload["rating"] = rating + return payload + + @dataclass(frozen=True) class PeerEntry: received_at: str @@ -154,6 +170,28 @@ def comment_body(self, peer_id: str, cid: str) -> str: raise_for_response(r) return r.text + def comment( + self, + peer_id: str, + *, + text: str, + rating: Optional[float] = None, + language: str = "en", + ) -> CommentSubmitResult: + """POST /v1/peers/{peer_id}/comments/self — leave a self-signed comment. + + The gateway signs the ledger entry with its own libp2p identity; the + SDK holds no keys. When ``rating`` (in [-1.0, +1.0]) is given, the boss + also appends a paired honesty Rating entry. + """ + r = self._http.post( + f"/v1/peers/{peer_id}/comments/self", + json=_comment_payload(text, language, rating), + ) + raise_for_response(r) + body = r.json() + return CommentSubmitResult(cid=body["cid"], ledger_hash=body["ledger_hash"]) + class AsyncPeersNamespace: """``client.peers.*`` — async peer discovery and trust inspection (v1.3.1).""" @@ -189,9 +227,27 @@ async def comment_body(self, peer_id: str, cid: str) -> str: raise_for_response(r) return r.text + async def comment( + self, + peer_id: str, + *, + text: str, + rating: Optional[float] = None, + language: str = "en", + ) -> CommentSubmitResult: + """POST /v1/peers/{peer_id}/comments/self — leave a self-signed comment.""" + r = await self._http.post( + f"/v1/peers/{peer_id}/comments/self", + json=_comment_payload(text, language, rating), + ) + raise_for_response(r) + body = r.json() + return CommentSubmitResult(cid=body["cid"], ledger_hash=body["ledger_hash"]) + __all__ = [ "AsyncPeersNamespace", + "CommentSubmitResult", "KnownPeer", "PeerEntry", "PeerSummary", diff --git a/agentfm-python/tests/unit/test_peers.py b/agentfm-python/tests/unit/test_peers.py index f34140f..9c06f1d 100644 --- a/agentfm-python/tests/unit/test_peers.py +++ b/agentfm-python/tests/unit/test_peers.py @@ -254,11 +254,59 @@ def test_peers_comment_body_returns_unicode(): assert "—" in body +# --------------------------------------------------------------------------- +# Sync: comment (self-signed submission) +# --------------------------------------------------------------------------- + + +def test_peers_comment_submits_and_returns_result(): + with respx.mock(base_url=GATEWAY, assert_all_called=True) as router: + route = router.post("/v1/peers/12D3KooWA/comments/self").mock( + return_value=Response(201, json={"cid": "1220abc", "ledger_hash": "deadbeef"}) + ) + with AgentFMClient(gateway_url=GATEWAY) as c: + res = c.peers.comment("12D3KooWA", text="Solid work", rating=0.6) + + assert res.cid == "1220abc" + assert res.ledger_hash == "deadbeef" + import json as _json + + sent = _json.loads(route.calls.last.request.content) + assert sent["text"] == "Solid work" + assert sent["rating"] == 0.6 + + +def test_peers_comment_omits_rating_when_none(): + with respx.mock(base_url=GATEWAY, assert_all_called=True) as router: + route = router.post("/v1/peers/12D3KooWA/comments/self").mock( + return_value=Response(201, json={"cid": "1220x", "ledger_hash": "h"}) + ) + with AgentFMClient(gateway_url=GATEWAY) as c: + c.peers.comment("12D3KooWA", text="No rating") + + import json as _json + + sent = _json.loads(route.calls.last.request.content) + assert "rating" not in sent + + # --------------------------------------------------------------------------- # Async: mirrors of sync tests # --------------------------------------------------------------------------- +async def test_async_peers_comment_submits_and_returns_result(): + with respx.mock(base_url=GATEWAY, assert_all_called=True) as router: + router.post("/v1/peers/12D3KooWA/comments/self").mock( + return_value=Response(201, json={"cid": "1220y", "ledger_hash": "z"}) + ) + async with AsyncAgentFMClient(gateway_url=GATEWAY) as c: + res = await c.peers.comment("12D3KooWA", text="Async solid", rating=-0.3) + + assert res.cid == "1220y" + assert res.ledger_hash == "z" + + async def test_async_peers_list_returns_online_and_offline(): with respx.mock(base_url=GATEWAY, assert_all_called=True) as router: router.get("/api/workers").mock( diff --git a/docs/superpowers/plans/2026-05-23-metrics-dashboard-implementation.md b/docs/superpowers/plans/2026-05-23-metrics-dashboard-implementation.md deleted file mode 100644 index 215e6a2..0000000 --- a/docs/superpowers/plans/2026-05-23-metrics-dashboard-implementation.md +++ /dev/null @@ -1,1911 +0,0 @@ -# Metrics Dashboard Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Add a real-time `/dashboard` route to the AgentFM desktop app powered by the boss `/metrics` endpoint, plus a per-worker `` on PeerView powered by the existing `/api/workers` poll. Renderer-only — zero Go or Electron-main changes. - -**Architecture:** Renderer polls `http://127.0.0.1:8080/metrics` every 2 s, parses Prometheus text into samples, pushes each series into a Zustand-backed ring buffer (5 min × 2 s = 150 points). A separate hook taps the already-running `useWorkers` React Query and appends per-peer telemetry to a parallel buffer. Charts read from the store: `uPlot` for the dashboard, hand-rolled canvas `SparkLine` for the strip. - -**Tech Stack:** TypeScript, React 18, Zustand (already in deps), React Query (already), uPlot (new dep), Vitest, Playwright. - -**Spec:** `docs/superpowers/specs/2026-05-23-metrics-dashboard-design.md` - -**Working directory for all `npm` / `npx` / `git` commands below:** `/Users/saif/Desktop/agentfm-prod/agentfm-desktop` - ---- - -## Task 1: Add the `uplot` dependency - -**Files:** -- Modify: `package.json` (auto-updated by npm) - -- [ ] **Step 1: Install uplot** - -```bash -cd /Users/saif/Desktop/agentfm-prod/agentfm-desktop -npm install uplot -``` - -Expected output: `added 1 package`. `package.json` and `package-lock.json` updated. - -- [ ] **Step 2: Verify the version landed in deps** - -```bash -node -e "console.log(require('./package.json').dependencies.uplot)" -``` - -Expected: prints a version string like `^1.6.30`. - -- [ ] **Step 3: Commit** - -```bash -git add package.json package-lock.json -git commit -m "chore(desktop): add uplot dependency for metrics charts" -``` - ---- - -## Task 2: Define metrics types - -**Files:** -- Create: `src/types/metrics.ts` - -- [ ] **Step 1: Create the type module** - -```ts -// src/types/metrics.ts - -export type MetricType = 'counter' | 'gauge' | 'histogram' | 'summary' | 'unknown' - -export interface MetricSample { - name: string - labels: Record - value: number - type: MetricType -} - -export interface RingBuffer { - ts: Float64Array - v: Float64Array - head: number - filled: number -} - -export const RING_CAPACITY = 150 - -export function createRingBuffer(): RingBuffer { - return { - ts: new Float64Array(RING_CAPACITY), - v: new Float64Array(RING_CAPACITY), - head: 0, - filled: 0, - } -} - -export function pushRing(buf: RingBuffer, ts: number, v: number): void { - buf.ts[buf.head] = ts - buf.v[buf.head] = v - buf.head = (buf.head + 1) % RING_CAPACITY - if (buf.filled < RING_CAPACITY) buf.filled++ -} - -export function latestValue(buf: RingBuffer): number | undefined { - if (buf.filled === 0) return undefined - const idx = (buf.head - 1 + RING_CAPACITY) % RING_CAPACITY - return buf.v[idx] -} - -export function ringToArrays(buf: RingBuffer): { ts: number[]; v: number[] } { - if (buf.filled === 0) return { ts: [], v: [] } - const ts: number[] = [] - const v: number[] = [] - const start = buf.filled < RING_CAPACITY ? 0 : buf.head - for (let i = 0; i < buf.filled; i++) { - const idx = (start + i) % RING_CAPACITY - ts.push(buf.ts[idx]) - v.push(buf.v[idx]) - } - return { ts, v } -} -``` - -- [ ] **Step 2: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. - -- [ ] **Step 3: Commit** - -```bash -git add src/types/metrics.ts -git commit -m "feat(desktop): add MetricSample type and ring-buffer primitives" -``` - ---- - -## Task 3: Implement ring buffer tests - -**Files:** -- Create: `tests/unit/ringBuffer.test.ts` - -- [ ] **Step 1: Write the failing tests** - -```ts -// tests/unit/ringBuffer.test.ts -import { describe, it, expect } from 'vitest' -import { - createRingBuffer, - pushRing, - latestValue, - ringToArrays, - RING_CAPACITY, -} from '../../src/types/metrics' - -describe('RingBuffer', () => { - it('starts empty', () => { - const b = createRingBuffer() - expect(b.filled).toBe(0) - expect(latestValue(b)).toBeUndefined() - expect(ringToArrays(b)).toEqual({ ts: [], v: [] }) - }) - - it('pushes one value', () => { - const b = createRingBuffer() - pushRing(b, 100, 7) - expect(b.filled).toBe(1) - expect(latestValue(b)).toBe(7) - expect(ringToArrays(b)).toEqual({ ts: [100], v: [7] }) - }) - - it('preserves insertion order before wrap', () => { - const b = createRingBuffer() - for (let i = 0; i < 10; i++) pushRing(b, i, i * 2) - const { ts, v } = ringToArrays(b) - expect(ts).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) - expect(v).toEqual([0, 2, 4, 6, 8, 10, 12, 14, 16, 18]) - expect(latestValue(b)).toBe(18) - }) - - it('wraps at capacity and drops oldest', () => { - const b = createRingBuffer() - for (let i = 0; i < RING_CAPACITY + 5; i++) pushRing(b, i, i) - const { ts, v } = ringToArrays(b) - expect(ts.length).toBe(RING_CAPACITY) - expect(v.length).toBe(RING_CAPACITY) - expect(ts[0]).toBe(5) // oldest is push #5 (push 0..4 were overwritten) - expect(v[v.length - 1]).toBe(RING_CAPACITY + 4) // newest - expect(latestValue(b)).toBe(RING_CAPACITY + 4) - }) - - it('latestValue reflects most recent push after wrap', () => { - const b = createRingBuffer() - for (let i = 0; i < RING_CAPACITY * 2; i++) pushRing(b, i, i) - expect(latestValue(b)).toBe(RING_CAPACITY * 2 - 1) - }) -}) -``` - -- [ ] **Step 2: Run — expect pass (no logic to write, but tests must pass)** - -```bash -npm test -- ringBuffer -``` - -Expected: 5 passing tests. If anything fails, fix `src/types/metrics.ts` from Task 2 before continuing. - -- [ ] **Step 3: Commit** - -```bash -git add tests/unit/ringBuffer.test.ts -git commit -m "test(desktop): cover ring-buffer push/wrap/latest semantics" -``` - ---- - -## Task 4: Implement the Prometheus parser (test-first) - -**Files:** -- Create: `tests/unit/fixtures/metrics-sample.txt` -- Create: `tests/unit/promParse.test.ts` -- Create: `src/lib/promParse.ts` - -- [ ] **Step 1: Write the fixture (representative `/metrics` snippet)** - -``` -# HELP agentfm_tasks_total Number of task executions, partitioned by terminal status. -# TYPE agentfm_tasks_total counter -agentfm_tasks_total{status="error"} 0 -agentfm_tasks_total{status="ok"} 142 -agentfm_tasks_total{status="rejected"} 0 -agentfm_tasks_total{status="timeout"} 0 -# HELP agentfm_task_duration_seconds Wall-clock task duration in seconds. -# TYPE agentfm_task_duration_seconds histogram -agentfm_task_duration_seconds_bucket{le="1"} 12 -agentfm_task_duration_seconds_bucket{le="5"} 45 -agentfm_task_duration_seconds_bucket{le="15"} 92 -agentfm_task_duration_seconds_bucket{le="60"} 128 -agentfm_task_duration_seconds_bucket{le="+Inf"} 142 -agentfm_task_duration_seconds_sum 1234.5 -agentfm_task_duration_seconds_count 142 -# HELP agentfm_workers_online Number of workers currently visible in this node's telemetry. -# TYPE agentfm_workers_online gauge -agentfm_workers_online 5 -# HELP agentfm_stream_errors_total Stream-level failures on AgentFM libp2p protocols. -# TYPE agentfm_stream_errors_total counter -agentfm_stream_errors_total{protocol="task",reason="deadline"} 2 -agentfm_stream_errors_total{protocol="task",reason="reset"} 0 -agentfm_stream_errors_total{protocol="artifacts",reason="peer_eof"} 1 -# HELP process_resident_memory_bytes Resident memory size in bytes. -# TYPE process_resident_memory_bytes gauge -process_resident_memory_bytes 1.4892e+08 -# HELP go_goroutines Number of goroutines that currently exist. -# TYPE go_goroutines gauge -go_goroutines 87 -``` - -Write that exact content to `tests/unit/fixtures/metrics-sample.txt`. - -- [ ] **Step 2: Write the failing tests** - -```ts -// tests/unit/promParse.test.ts -import { describe, it, expect } from 'vitest' -import { readFileSync } from 'node:fs' -import { resolve } from 'node:path' -import { parseMetrics } from '../../src/lib/promParse' - -const FIXTURE = readFileSync( - resolve(__dirname, 'fixtures/metrics-sample.txt'), - 'utf8', -) - -describe('parseMetrics', () => { - it('returns [] for empty input', () => { - expect(parseMetrics('')).toEqual([]) - }) - - it('skips comments and empty lines', () => { - const samples = parseMetrics('# HELP foo bar\n# TYPE foo counter\n\nfoo 1') - expect(samples).toHaveLength(1) - expect(samples[0]).toMatchObject({ name: 'foo', value: 1, type: 'counter' }) - }) - - it('parses unlabeled gauges', () => { - const samples = parseMetrics('# TYPE x gauge\nx 5') - expect(samples[0]).toMatchObject({ name: 'x', value: 5, type: 'gauge', labels: {} }) - }) - - it('parses labels including multiple key/value pairs', () => { - const samples = parseMetrics( - '# TYPE e counter\ne{protocol="task",reason="reset"} 7', - ) - expect(samples[0].labels).toEqual({ protocol: 'task', reason: 'reset' }) - expect(samples[0].value).toBe(7) - }) - - it('parses scientific notation', () => { - const samples = parseMetrics('# TYPE m gauge\nm 1.4892e+08') - expect(samples[0].value).toBeCloseTo(1.4892e8) - }) - - it('parses +Inf as Infinity', () => { - const samples = parseMetrics( - '# TYPE h histogram\nh_bucket{le="+Inf"} 142', - ) - expect(samples[0].labels.le).toBe('+Inf') - expect(samples[0].value).toBe(142) - }) - - it('skips malformed lines without aborting', () => { - const text = '# TYPE a counter\na 1\nthis is not a valid line\na 2' - const samples = parseMetrics(text) - expect(samples.map((s) => s.value)).toEqual([1, 2]) - }) - - it('parses the full /metrics fixture', () => { - const samples = parseMetrics(FIXTURE) - const names = new Set(samples.map((s) => s.name)) - expect(names.has('agentfm_tasks_total')).toBe(true) - expect(names.has('agentfm_task_duration_seconds_bucket')).toBe(true) - expect(names.has('agentfm_task_duration_seconds_sum')).toBe(true) - expect(names.has('agentfm_task_duration_seconds_count')).toBe(true) - expect(names.has('agentfm_workers_online')).toBe(true) - expect(names.has('agentfm_stream_errors_total')).toBe(true) - expect(names.has('process_resident_memory_bytes')).toBe(true) - expect(names.has('go_goroutines')).toBe(true) - - const ok = samples.find( - (s) => s.name === 'agentfm_tasks_total' && s.labels.status === 'ok', - ) - expect(ok?.value).toBe(142) - - const online = samples.find((s) => s.name === 'agentfm_workers_online') - expect(online?.value).toBe(5) - expect(online?.type).toBe('gauge') - - const bucket = samples.find( - (s) => - s.name === 'agentfm_task_duration_seconds_bucket' && - s.labels.le === '60', - ) - expect(bucket?.value).toBe(128) - expect(bucket?.type).toBe('histogram') - }) -}) -``` - -- [ ] **Step 3: Run tests — expect failure (parser doesn't exist yet)** - -```bash -npm test -- promParse -``` - -Expected: failure with "Cannot find module './src/lib/promParse'" or similar. - -- [ ] **Step 4: Implement the parser** - -```ts -// src/lib/promParse.ts -import type { MetricSample, MetricType } from '../types/metrics' - -const TYPE_RE = /^#\s*TYPE\s+(\S+)\s+(counter|gauge|histogram|summary)\s*$/ -const SAMPLE_RE = /^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+(-?[\d.eE+-]+|NaN|\+Inf|-Inf)\s*$/ -const LABEL_RE = /([a-zA-Z_][a-zA-Z0-9_]*)="((?:[^"\\]|\\.)*)"/g - -function parseLabels(raw: string | undefined): Record { - if (!raw) return {} - const out: Record = {} - LABEL_RE.lastIndex = 0 - let m: RegExpExecArray | null - while ((m = LABEL_RE.exec(raw)) !== null) { - out[m[1]] = m[2].replace(/\\(.)/g, '$1') - } - return out -} - -function parseValue(raw: string): number { - if (raw === '+Inf') return Infinity - if (raw === '-Inf') return -Infinity - if (raw === 'NaN') return NaN - return Number(raw) -} - -function baseName(name: string): string { - // Histogram suffixes (_bucket / _sum / _count) and summary suffixes share - // the parent metric's declared TYPE. Strip suffix to look up the TYPE entry. - if (name.endsWith('_bucket')) return name.slice(0, -'_bucket'.length) - if (name.endsWith('_sum')) return name.slice(0, -'_sum'.length) - if (name.endsWith('_count')) return name.slice(0, -'_count'.length) - return name -} - -export function parseMetrics(text: string): MetricSample[] { - const out: MetricSample[] = [] - const types = new Map() - const lines = text.split('\n') - for (const rawLine of lines) { - const line = rawLine.trim() - if (!line) continue - try { - if (line.startsWith('#')) { - const t = line.match(TYPE_RE) - if (t) types.set(t[1], t[2] as MetricType) - continue - } - const m = line.match(SAMPLE_RE) - if (!m) continue - const name = m[1] - const labels = parseLabels(m[2]) - const value = parseValue(m[3]) - if (!Number.isFinite(value) && !Number.isNaN(value)) { - // Allow +Inf for histogram buckets; otherwise treat as numeric. - } - const type = types.get(baseName(name)) ?? 'unknown' - out.push({ name, labels, value, type }) - } catch { - // Defensive: malformed individual lines must not abort the rest of - // the parse. Silently skip. - } - } - return out -} -``` - -- [ ] **Step 5: Run tests — expect pass** - -```bash -npm test -- promParse -``` - -Expected: 8 passing tests. - -- [ ] **Step 6: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. - -- [ ] **Step 7: Commit** - -```bash -git add tests/unit/fixtures/metrics-sample.txt tests/unit/promParse.test.ts src/lib/promParse.ts -git commit -m "feat(desktop): parse Prometheus text into MetricSample[]" -``` - ---- - -## Task 5: Implement the metrics store (Zustand) - -**Files:** -- Create: `src/lib/metricsStore.ts` -- Create: `tests/unit/metricsStore.test.ts` - -- [ ] **Step 1: Write the failing tests** - -```ts -// tests/unit/metricsStore.test.ts -import { describe, it, expect, beforeEach } from 'vitest' -import { useMetricsStore, seriesKey } from '../../src/lib/metricsStore' -import { latestValue, ringToArrays } from '../../src/types/metrics' - -beforeEach(() => { - useMetricsStore.getState().reset() -}) - -describe('metricsStore.pushBoss', () => { - it('creates buffers on first push', () => { - useMetricsStore.getState().pushBoss(1000, [ - { name: 'agentfm_tasks_total', labels: { status: 'ok' }, value: 5, type: 'counter' }, - ]) - const key = seriesKey('agentfm_tasks_total', { status: 'ok' }) - const buf = useMetricsStore.getState().bossSeries.get(key) - expect(buf).toBeDefined() - expect(latestValue(buf!)).toBe(5) - }) - - it('appends multiple ticks to the same series', () => { - const s = useMetricsStore.getState() - s.pushBoss(1000, [ - { name: 'g', labels: {}, value: 1, type: 'gauge' }, - ]) - s.pushBoss(2000, [ - { name: 'g', labels: {}, value: 2, type: 'gauge' }, - ]) - s.pushBoss(3000, [ - { name: 'g', labels: {}, value: 3, type: 'gauge' }, - ]) - const buf = s.bossSeries.get(seriesKey('g', {}))! - expect(ringToArrays(buf)).toEqual({ ts: [1000, 2000, 3000], v: [1, 2, 3] }) - }) - - it('carries forward series that are missing this tick', () => { - const s = useMetricsStore.getState() - s.pushBoss(1000, [ - { name: 'a', labels: {}, value: 10, type: 'gauge' }, - { name: 'b', labels: {}, value: 20, type: 'gauge' }, - ]) - s.pushBoss(2000, [ - { name: 'a', labels: {}, value: 11, type: 'gauge' }, - // 'b' absent this tick - ]) - const b = s.bossSeries.get(seriesKey('b', {}))! - expect(ringToArrays(b)).toEqual({ ts: [1000, 2000], v: [20, 20] }) - }) -}) - -describe('metricsStore.pushPeer', () => { - it('isolates per-peer buffers', () => { - const s = useMetricsStore.getState() - s.pushPeer('peerA', 1000, { cpu: 50, gpu: 0, ram: 4, queue: 1 }) - s.pushPeer('peerB', 1000, { cpu: 90, gpu: 0, ram: 2, queue: 3 }) - const a = s.peerSeries.get('peerA')! - const b = s.peerSeries.get('peerB')! - expect(latestValue(a.get('cpu')!)).toBe(50) - expect(latestValue(b.get('cpu')!)).toBe(90) - expect(latestValue(a.get('queue')!)).toBe(1) - expect(latestValue(b.get('queue')!)).toBe(3) - }) -}) - -describe('seriesKey', () => { - it('produces stable keys regardless of label insertion order', () => { - const k1 = seriesKey('m', { a: '1', b: '2' }) - const k2 = seriesKey('m', { b: '2', a: '1' }) - expect(k1).toBe(k2) - }) -}) -``` - -- [ ] **Step 2: Run tests — expect failure** - -```bash -npm test -- metricsStore -``` - -Expected: module not found. - -- [ ] **Step 3: Implement the store** - -```ts -// src/lib/metricsStore.ts -import { create } from 'zustand' -import { - createRingBuffer, - pushRing, - latestValue, - RingBuffer, -} from '../types/metrics' -import type { MetricSample } from '../types/metrics' - -export type PeerMetric = 'cpu' | 'gpu' | 'ram' | 'queue' - -export interface PeerSnapshot { - cpu: number - gpu: number - ram: number - queue: number -} - -interface MetricsState { - bossSeries: Map - peerSeries: Map> - peerLastTick: Map - lastBossTick: number - pushBoss: (ts: number, samples: MetricSample[]) => void - pushPeer: (peerId: string, ts: number, snap: PeerSnapshot) => void - reset: () => void -} - -export function seriesKey(name: string, labels: Record): string { - const keys = Object.keys(labels).sort() - if (keys.length === 0) return name - const pairs = keys.map((k) => `${k}=${labels[k]}`).join(',') - return `${name}{${pairs}}` -} - -export const useMetricsStore = create((set, get) => ({ - bossSeries: new Map(), - peerSeries: new Map(), - peerLastTick: new Map(), - lastBossTick: 0, - - pushBoss: (ts, samples) => { - const { bossSeries } = get() - const seen = new Set() - for (const s of samples) { - const k = seriesKey(s.name, s.labels) - let buf = bossSeries.get(k) - if (!buf) { - buf = createRingBuffer() - bossSeries.set(k, buf) - } - pushRing(buf, ts, s.value) - seen.add(k) - } - // Carry-forward: any series that already had data but didn't get a - // sample this tick gets its previous value pushed with the new ts. - for (const [k, buf] of bossSeries) { - if (seen.has(k)) continue - const last = latestValue(buf) - if (last !== undefined) pushRing(buf, ts, last) - } - // New Map reference triggers Zustand subscribers. - set({ bossSeries: new Map(bossSeries), lastBossTick: ts }) - }, - - pushPeer: (peerId, ts, snap) => { - const { peerSeries, peerLastTick } = get() - let peerBufs = peerSeries.get(peerId) - if (!peerBufs) { - peerBufs = new Map([ - ['cpu', createRingBuffer()], - ['gpu', createRingBuffer()], - ['ram', createRingBuffer()], - ['queue', createRingBuffer()], - ]) - peerSeries.set(peerId, peerBufs) - } - pushRing(peerBufs.get('cpu')!, ts, snap.cpu) - pushRing(peerBufs.get('gpu')!, ts, snap.gpu) - pushRing(peerBufs.get('ram')!, ts, snap.ram) - pushRing(peerBufs.get('queue')!, ts, snap.queue) - peerLastTick.set(peerId, ts) - set({ - peerSeries: new Map(peerSeries), - peerLastTick: new Map(peerLastTick), - }) - }, - - reset: () => - set({ - bossSeries: new Map(), - peerSeries: new Map(), - peerLastTick: new Map(), - lastBossTick: 0, - }), -})) -``` - -- [ ] **Step 4: Run tests — expect pass** - -```bash -npm test -- metricsStore -``` - -Expected: all 5 passing. - -- [ ] **Step 5: Commit** - -```bash -git add src/lib/metricsStore.ts tests/unit/metricsStore.test.ts -git commit -m "feat(desktop): zustand metrics store with carry-forward ring buffers" -``` - ---- - -## Task 6: Implement derived computations - -**Files:** -- Create: `src/lib/metricsDerive.ts` -- Create: `tests/unit/metricsDerive.test.ts` - -- [ ] **Step 1: Write the failing tests** - -```ts -// tests/unit/metricsDerive.test.ts -import { describe, it, expect } from 'vitest' -import { - computeRate, - computeTasksPerMinute, - computeP95FromBuckets, -} from '../../src/lib/metricsDerive' -import { createRingBuffer, pushRing } from '../../src/types/metrics' - -describe('computeRate', () => { - it('returns 0 when buffer has fewer than 2 points', () => { - const b = createRingBuffer() - expect(computeRate(b)).toBe(0) - pushRing(b, 1000, 5) - expect(computeRate(b)).toBe(0) - }) - - it('computes per-second rate between first and last samples', () => { - const b = createRingBuffer() - pushRing(b, 0, 0) - pushRing(b, 1000, 10) - pushRing(b, 2000, 30) - // (30 - 0) / 2s = 15/s - expect(computeRate(b)).toBeCloseTo(15) - }) - - it('returns 0 when timestamps collide (avoid div-by-zero)', () => { - const b = createRingBuffer() - pushRing(b, 1000, 5) - pushRing(b, 1000, 10) - expect(computeRate(b)).toBe(0) - }) - - it('clamps negative rates (counter reset) to 0', () => { - const b = createRingBuffer() - pushRing(b, 0, 100) - pushRing(b, 1000, 5) - expect(computeRate(b)).toBe(0) - }) -}) - -describe('computeTasksPerMinute', () => { - it('returns 0 on empty buffer', () => { - expect(computeTasksPerMinute(createRingBuffer())).toBe(0) - }) - - it('returns rate × 60', () => { - const b = createRingBuffer() - pushRing(b, 0, 0) - pushRing(b, 1000, 1) // 1 task/s = 60/min - expect(computeTasksPerMinute(b)).toBeCloseTo(60) - }) -}) - -describe('computeP95FromBuckets', () => { - it('returns 0 for empty buckets', () => { - expect(computeP95FromBuckets([])).toBe(0) - }) - - it('returns 0 when total count is 0', () => { - expect(computeP95FromBuckets([{ le: 1, count: 0 }, { le: Infinity, count: 0 }])).toBe(0) - }) - - it('interpolates p95 within the right bucket', () => { - // 100 total. p95 = 95th. Buckets: ≤1s=10, ≤5s=80, ≤15s=95, ≤60s=98, ≤Inf=100 - const p95 = computeP95FromBuckets([ - { le: 1, count: 10 }, - { le: 5, count: 80 }, - { le: 15, count: 95 }, - { le: 60, count: 98 }, - { le: Infinity, count: 100 }, - ]) - // p95 falls exactly at the boundary of ≤15. Linear interp inside the - // (5, 15] bucket: 95 - 80 = 15 of 15 cumulative = full bucket → 15. - expect(p95).toBeCloseTo(15) - }) - - it('returns the highest finite bucket when p95 lands in +Inf', () => { - const p95 = computeP95FromBuckets([ - { le: 1, count: 10 }, - { le: 60, count: 50 }, - { le: Infinity, count: 100 }, - ]) - // p95 = 95, falls in (60, +Inf] — report 60 as a finite upper bound. - expect(p95).toBe(60) - }) -}) -``` - -- [ ] **Step 2: Run — expect failure** - -```bash -npm test -- metricsDerive -``` - -Expected: module not found. - -- [ ] **Step 3: Implement** - -```ts -// src/lib/metricsDerive.ts -import type { RingBuffer } from '../types/metrics' -import { latestValue, ringToArrays } from '../types/metrics' - -export function computeRate(buf: RingBuffer): number { - if (buf.filled < 2) return 0 - const { ts, v } = ringToArrays(buf) - const dt = (ts[ts.length - 1] - ts[0]) / 1000 - if (dt <= 0) return 0 - const dv = v[v.length - 1] - v[0] - if (dv < 0) return 0 - return dv / dt -} - -export function computeTasksPerMinute(buf: RingBuffer): number { - return computeRate(buf) * 60 -} - -export interface HistogramBucket { - le: number - count: number -} - -export function computeP95FromBuckets(buckets: HistogramBucket[]): number { - if (buckets.length === 0) return 0 - // Buckets arrive in ascending `le` order; the last finite or +Inf bucket - // holds the total count. - const sorted = [...buckets].sort((a, b) => a.le - b.le) - const total = sorted[sorted.length - 1].count - if (total <= 0) return 0 - const target = total * 0.95 - let prevLe = 0 - let prevCount = 0 - for (const b of sorted) { - if (b.count >= target) { - if (!Number.isFinite(b.le)) { - // p95 lands in the +Inf bucket — return the highest finite le. - const lastFinite = sorted - .filter((x) => Number.isFinite(x.le)) - .map((x) => x.le) - .pop() - return lastFinite ?? 0 - } - const bucketSize = b.count - prevCount - if (bucketSize <= 0) return b.le - const frac = (target - prevCount) / bucketSize - return prevLe + frac * (b.le - prevLe) - } - prevLe = b.le - prevCount = b.count - } - return sorted[sorted.length - 1].le -} - -export { latestValue } -``` - -- [ ] **Step 4: Run — expect pass** - -```bash -npm test -- metricsDerive -``` - -Expected: 8 passing. - -- [ ] **Step 5: Commit** - -```bash -git add src/lib/metricsDerive.ts tests/unit/metricsDerive.test.ts -git commit -m "feat(desktop): derive rate, tasks/min, and p95 from histogram buckets" -``` - ---- - -## Task 7: Implement the `/metrics` polling hook - -**Files:** -- Create: `src/hooks/useMetricsPoll.ts` - -- [ ] **Step 1: Implement the hook** - -```ts -// src/hooks/useMetricsPoll.ts -import { useEffect, useRef } from 'react' -import { getApiBaseURL } from '../lib/api' -import { parseMetrics } from '../lib/promParse' -import { useMetricsStore } from '../lib/metricsStore' - -const FAST_INTERVAL_MS = 2_000 -const SLOW_INTERVAL_MS = 10_000 -const ERROR_THRESHOLD = 3 - -/** - * Polls the boss /metrics endpoint while the document is visible and the - * hook is mounted. Pauses on `visibilitychange:hidden`. Switches to a 10s - * backoff after 3 consecutive errors; returns to 2s on first success. - * - * Call this from inside a route component that should drive polling (only - * Dashboard today). Other routes that don't need /metrics should not call - * it — the boss `/metrics` endpoint is unrelated to the always-on - * `/api/workers` poll handled by useWorkerHistory. - */ -export function useMetricsPoll(): void { - const errorsRef = useRef(0) - const cancelledRef = useRef(false) - const timerRef = useRef | null>(null) - const pushBoss = useMetricsStore((s) => s.pushBoss) - - useEffect(() => { - cancelledRef.current = false - - async function tick() { - if (cancelledRef.current) return - if (document.visibilityState !== 'visible') { - // Try again on next visibility change; do not schedule a timer. - return - } - try { - const res = await fetch(`${getApiBaseURL()}/metrics`) - if (!res.ok) throw new Error(`HTTP ${res.status}`) - const text = await res.text() - const samples = parseMetrics(text) - pushBoss(Date.now(), samples) - errorsRef.current = 0 - } catch { - errorsRef.current++ - } - if (cancelledRef.current) return - const next = - errorsRef.current >= ERROR_THRESHOLD ? SLOW_INTERVAL_MS : FAST_INTERVAL_MS - timerRef.current = setTimeout(tick, next) - } - - function onVisibility() { - if (document.visibilityState === 'visible' && !timerRef.current) { - tick() - } - } - - document.addEventListener('visibilitychange', onVisibility) - tick() - - return () => { - cancelledRef.current = true - if (timerRef.current) clearTimeout(timerRef.current) - timerRef.current = null - document.removeEventListener('visibilitychange', onVisibility) - } - }, [pushBoss]) -} -``` - -- [ ] **Step 2: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. - -- [ ] **Step 3: Commit** - -```bash -git add src/hooks/useMetricsPoll.ts -git commit -m "feat(desktop): 2s metrics poll with visibility-pause and backoff" -``` - ---- - -## Task 8: Implement the per-worker history hook - -**Files:** -- Create: `src/hooks/useWorkerHistory.ts` - -- [ ] **Step 1: Implement** - -```ts -// src/hooks/useWorkerHistory.ts -import { useEffect } from 'react' -import { useWorkers } from '../lib/query' -import { useMetricsStore } from '../lib/metricsStore' -import type { WorkerProfile } from '../types/api' - -/** - * Continuously captures a per-peer history of CPU%, GPU%, RAM-free (GB), - * and queue depth from the existing useWorkers React Query poll. - * - * Call this from App.tsx so the buffer is always populated regardless of - * which route is visible — when a user navigates to PeerView, charts - * already have several minutes of history. - */ -export function useWorkerHistory(): void { - const { data } = useWorkers(true) // include offline so we keep buffers visible - const pushPeer = useMetricsStore((s) => s.pushPeer) - - useEffect(() => { - if (!data) return - const now = Date.now() - for (const w of data.agents as WorkerProfile[]) { - // Only push for peers that are currently online — offline peers' - // last_seen sample is already in the buffer from when they were live. - if (!w.online) continue - pushPeer(w.peer_id, now, { - cpu: w.cpu_usage_pct ?? 0, - gpu: w.gpu_usage_pct ?? 0, - ram: w.ram_free_gb ?? 0, - queue: w.current_tasks ?? 0, - }) - } - }, [data, pushPeer]) -} -``` - -- [ ] **Step 2: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. - -- [ ] **Step 3: Commit** - -```bash -git add src/hooks/useWorkerHistory.ts -git commit -m "feat(desktop): capture per-peer telemetry history from /api/workers poll" -``` - ---- - -## Task 9: Implement the canvas SparkLine - -**Files:** -- Create: `src/components/charts/SparkLine.tsx` -- Create: `tests/unit/sparkLine.test.tsx` - -- [ ] **Step 1: Write the smoke test** - -```tsx -// tests/unit/sparkLine.test.tsx -import { describe, it, expect } from 'vitest' -import { render } from '@testing-library/react' -import { SparkLine } from '../../src/components/charts/SparkLine' - -describe('SparkLine', () => { - it('renders a canvas with the right dimensions', () => { - const { container } = render( - , - ) - const canvas = container.querySelector('canvas') - expect(canvas).toBeTruthy() - expect(canvas!.getAttribute('width')).toBe('240') // 120 * dpr (defaults to 2 in jsdom) - }) - - it('renders an empty canvas for empty values', () => { - const { container } = render( - , - ) - const canvas = container.querySelector('canvas') - expect(canvas).toBeTruthy() - }) - - it('renders for a single value', () => { - const { container } = render( - , - ) - expect(container.querySelector('canvas')).toBeTruthy() - }) -}) -``` - -> **Note on jsdom dpr:** jsdom defaults `window.devicePixelRatio` to 1. The test asserts width = 120 if dpr=1, or 240 if dpr=2. If the assertion fails because of dpr, change `expect('240')` to `expect('120')`. The behaviour under test is "canvas gets a width attribute"; the exact value is implementation detail. - -- [ ] **Step 2: Run — expect failure** - -```bash -npm test -- sparkLine -``` - -Expected: module not found. - -- [ ] **Step 3: Implement** - -```tsx -// src/components/charts/SparkLine.tsx -import { useEffect, useRef } from 'react' - -export interface SparkLineProps { - values: number[] - width: number - height: number - color: string -} - -export function SparkLine({ values, width, height, color }: SparkLineProps) { - const ref = useRef(null) - - useEffect(() => { - const canvas = ref.current - if (!canvas) return - const ctx = canvas.getContext('2d') - if (!ctx) return - - const dpr = window.devicePixelRatio || 1 - canvas.width = width * dpr - canvas.height = height * dpr - canvas.style.width = `${width}px` - canvas.style.height = `${height}px` - ctx.scale(dpr, dpr) - ctx.clearRect(0, 0, width, height) - - if (values.length === 0) return - - let min = values[0] - let max = values[0] - for (const v of values) { - if (v < min) min = v - if (v > max) max = v - } - const range = max - min || 1 - const stepX = values.length > 1 ? width / (values.length - 1) : width - - ctx.beginPath() - for (let i = 0; i < values.length; i++) { - const x = i * stepX - const y = height - ((values[i] - min) / range) * (height - 2) - 1 - if (i === 0) ctx.moveTo(x, y) - else ctx.lineTo(x, y) - } - ctx.strokeStyle = color - ctx.lineWidth = 1.2 - ctx.stroke() - - // Soft fill under the line. - ctx.lineTo(width, height) - ctx.lineTo(0, height) - ctx.closePath() - ctx.fillStyle = color + '22' // hex + ~13% alpha - ctx.fill() - }, [values, width, height, color]) - - return -} -``` - -- [ ] **Step 4: Run — expect pass** - -```bash -npm test -- sparkLine -``` - -Expected: 3 passing. If `width` assertion fails because jsdom uses dpr=1, edit the test to expect `'120'`. - -- [ ] **Step 5: Commit** - -```bash -git add src/components/charts/SparkLine.tsx tests/unit/sparkLine.test.tsx -git commit -m "feat(desktop): canvas-based SparkLine for telemetry strips" -``` - ---- - -## Task 10: Implement the UPlotChart wrapper - -**Files:** -- Create: `src/components/charts/UPlotChart.tsx` -- Modify: `src/styles/global.css` (add uplot stylesheet import) - -- [ ] **Step 1: Locate the renderer's global stylesheet** - -```bash -ls /Users/saif/Desktop/agentfm-prod/agentfm-desktop/src/styles/ -``` - -Note the file name (likely `globals.css` or `index.css`). The next step imports the uPlot CSS once globally. - -- [ ] **Step 2: Add the uPlot CSS import** - -In the global stylesheet (e.g. `src/styles/globals.css`), add the following line at the top of the imports block: - -```css -@import 'uplot/dist/uPlot.min.css'; -``` - -- [ ] **Step 3: Implement the wrapper** - -```tsx -// src/components/charts/UPlotChart.tsx -import { useEffect, useRef } from 'react' -import uPlot from 'uplot' -import type { Options, AlignedData } from 'uplot' - -export interface UPlotChartProps { - data: AlignedData - height: number - series: { label: string; color: string }[] -} - -export function UPlotChart({ data, height, series }: UPlotChartProps) { - const containerRef = useRef(null) - const chartRef = useRef(null) - - useEffect(() => { - const el = containerRef.current - if (!el) return - - const opts: Options = { - width: el.clientWidth || 600, - height, - legend: { show: false }, - cursor: { show: false }, - scales: { x: { time: true } }, - axes: [ - { stroke: '#64748b', grid: { stroke: 'rgba(148,163,184,0.08)' } }, - { stroke: '#64748b', grid: { stroke: 'rgba(148,163,184,0.08)' } }, - ], - series: [ - {}, - ...series.map((s) => ({ - label: s.label, - stroke: s.color, - width: 1.5, - fill: s.color + '22', - })), - ], - } - - const chart = new uPlot(opts, data, el) - chartRef.current = chart - - const ro = new ResizeObserver(() => { - if (chartRef.current && el.clientWidth > 0) { - chartRef.current.setSize({ width: el.clientWidth, height }) - } - }) - ro.observe(el) - - return () => { - ro.disconnect() - chart.destroy() - chartRef.current = null - } - // The wrapper is intentionally re-created when `series` shape changes. - // setData below handles same-shape data updates without rebuild. - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [series.length, height]) - - useEffect(() => { - if (chartRef.current) { - chartRef.current.setData(data) - } - }, [data]) - - return
-} -``` - -- [ ] **Step 4: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. If `uplot` types are missing, install: `npm install --save-dev @types/uplot` and re-run. (`uplot` ships its own types since 1.6.x, but verify the version landed.) - -- [ ] **Step 5: Commit** - -```bash -git add src/components/charts/UPlotChart.tsx src/styles/ -git commit -m "feat(desktop): uPlot React wrapper with resize observer" -``` - ---- - -## Task 11: Implement the TelemetryStrip - -**Files:** -- Create: `src/components/peer/TelemetryStrip.tsx` -- Create: `tests/unit/telemetryStrip.test.tsx` - -- [ ] **Step 1: Write the smoke tests** - -```tsx -// tests/unit/telemetryStrip.test.tsx -import { describe, it, expect, beforeEach } from 'vitest' -import { render } from '@testing-library/react' -import { TelemetryStrip } from '../../src/components/peer/TelemetryStrip' -import { useMetricsStore } from '../../src/lib/metricsStore' - -beforeEach(() => { - useMetricsStore.getState().reset() -}) - -describe('TelemetryStrip', () => { - it('shows the waiting placeholder when no buffer exists', () => { - const { getByText } = render() - expect(getByText(/Waiting for telemetry beacon/i)).toBeTruthy() - }) - - it('renders sparkline cells when the buffer has data', () => { - useMetricsStore.getState().pushPeer('peerA', Date.now(), { - cpu: 32, - gpu: 68, - ram: 4.2, - queue: 2, - }) - const { container, getByText } = render() - expect(container.querySelectorAll('canvas').length).toBeGreaterThanOrEqual(4) - expect(getByText(/CPU/i)).toBeTruthy() - expect(getByText(/GPU/i)).toBeTruthy() - expect(getByText(/RAM/i)).toBeTruthy() - expect(getByText(/QUEUE/i)).toBeTruthy() - }) - - it('shows offline notice when last tick is > 30s ago', () => { - const longAgo = Date.now() - 60_000 - useMetricsStore.getState().pushPeer('peerB', longAgo, { - cpu: 10, gpu: 20, ram: 1, queue: 0, - }) - const { getByText } = render() - expect(getByText(/offline/i)).toBeTruthy() - }) -}) -``` - -- [ ] **Step 2: Run — expect failure** - -```bash -npm test -- telemetryStrip -``` - -Expected: module not found. - -- [ ] **Step 3: Implement** - -```tsx -// src/components/peer/TelemetryStrip.tsx -import { useMetricsStore } from '../../lib/metricsStore' -import { SparkLine } from '../charts/SparkLine' -import { latestValue, ringToArrays } from '../../types/metrics' -import type { PeerMetric } from '../../lib/metricsStore' - -const OFFLINE_AFTER_MS = 30_000 - -const CELLS: { metric: PeerMetric; label: string; color: string; fmt: (v: number) => string }[] = [ - { metric: 'cpu', label: 'CPU', color: '#22d3ee', fmt: (v) => `${Math.round(v)}%` }, - { metric: 'gpu', label: 'GPU', color: '#a855f7', fmt: (v) => `${Math.round(v)}%` }, - { metric: 'ram', label: 'RAM FREE', color: '#84cc16', fmt: (v) => `${v.toFixed(1)}G` }, - { metric: 'queue', label: 'QUEUE', color: '#f43f5e', fmt: (v) => `${Math.round(v)}` }, -] - -function formatAgo(ms: number): string { - if (ms < 60_000) return `${Math.round(ms / 1000)}s` - if (ms < 3_600_000) return `${Math.round(ms / 60_000)}m` - return `${Math.round(ms / 3_600_000)}h` -} - -export interface TelemetryStripProps { - peerId: string -} - -export function TelemetryStrip({ peerId }: TelemetryStripProps) { - const peerBufs = useMetricsStore((s) => s.peerSeries.get(peerId)) - const lastTick = useMetricsStore((s) => s.peerLastTick.get(peerId)) - - if (!peerBufs || !lastTick) { - return ( -
- Waiting for telemetry beacon… -
- ) - } - - const sinceLast = Date.now() - lastTick - const offline = sinceLast > OFFLINE_AFTER_MS - - return ( -
-
-
- - {offline ? `OFFLINE — last seen ${formatAgo(sinceLast)} ago` : 'LIVE TELEMETRY · LAST 5 MIN'} -
-
-
- {CELLS.map((c) => { - const buf = peerBufs.get(c.metric)! - const v = latestValue(buf) ?? 0 - const { v: values } = ringToArrays(buf) - return ( -
-
- {c.label} -
-
- {c.fmt(v)} -
- -
- ) - })} -
-
- ) -} -``` - -- [ ] **Step 4: Run — expect pass** - -```bash -npm test -- telemetryStrip -``` - -Expected: 3 passing. - -- [ ] **Step 5: Commit** - -```bash -git add src/components/peer/TelemetryStrip.tsx tests/unit/telemetryStrip.test.tsx -git commit -m "feat(desktop): TelemetryStrip with 4 sparklines and offline detection" -``` - ---- - -## Task 12: Implement the Dashboard route - -**Files:** -- Create: `src/routes/Dashboard.tsx` -- Create: `tests/unit/dashboard.test.tsx` - -- [ ] **Step 1: Write the smoke tests** - -```tsx -// tests/unit/dashboard.test.tsx -import { describe, it, expect, beforeEach, vi } from 'vitest' -import { render } from '@testing-library/react' -import Dashboard from '../../src/routes/Dashboard' -import { useMetricsStore } from '../../src/lib/metricsStore' - -// useMetricsPoll uses fetch + setInterval; stub fetch so tests don't hit -// the network. The hook returns void; we just need it to not throw. -beforeEach(() => { - useMetricsStore.getState().reset() - vi.stubGlobal( - 'fetch', - vi.fn().mockResolvedValue({ ok: true, text: () => Promise.resolve('') }), - ) -}) - -describe('Dashboard', () => { - it('renders without crashing on empty store', () => { - const { container } = render() - expect(container.textContent).toMatch(/TASKS/i) - }) - - it('renders the hero tile with the latest task count when data is seeded', () => { - useMetricsStore.getState().pushBoss(Date.now() - 1000, [ - { name: 'agentfm_tasks_total', labels: { status: 'ok' }, value: 100, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'error' }, value: 2, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'rejected' }, value: 0, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'timeout' }, value: 0, type: 'counter' }, - ]) - useMetricsStore.getState().pushBoss(Date.now(), [ - { name: 'agentfm_tasks_total', labels: { status: 'ok' }, value: 142, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'error' }, value: 3, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'rejected' }, value: 0, type: 'counter' }, - { name: 'agentfm_tasks_total', labels: { status: 'timeout' }, value: 1, type: 'counter' }, - ]) - const { container } = render() - // 142 + 3 + 0 + 1 = 146 - expect(container.textContent).toMatch(/146/) - }) -}) -``` - -- [ ] **Step 2: Run — expect failure** - -```bash -npm test -- dashboard -``` - -Expected: module not found. - -- [ ] **Step 3: Implement** - -```tsx -// src/routes/Dashboard.tsx -import { useMemo } from 'react' -import { useMetricsPoll } from '../hooks/useMetricsPoll' -import { useMetricsStore, seriesKey } from '../lib/metricsStore' -import { - computeRate, - computeTasksPerMinute, - computeP95FromBuckets, -} from '../lib/metricsDerive' -import { createRingBuffer, latestValue, ringToArrays } from '../types/metrics' -import { SparkLine } from '../components/charts/SparkLine' -import { SectionLabel } from '../components/primitives/SectionLabel' -import { HeroTitle } from '../components/primitives/HeroTitle' -import { NeonCard } from '../components/primitives/NeonCard' - -const STATUSES = ['ok', 'error', 'rejected', 'timeout'] as const -const STATUS_COLOR: Record<(typeof STATUSES)[number], string> = { - ok: '#84cc16', - error: '#f43f5e', - rejected: '#a855f7', - timeout: '#fbbf24', -} - -const STALE_MS = 10_000 - -export default function Dashboard() { - useMetricsPoll() - const bossSeries = useMetricsStore((s) => s.bossSeries) - const lastTick = useMetricsStore((s) => s.lastBossTick) - - const staleAgeMs = lastTick === 0 ? 0 : Date.now() - lastTick - const stale = lastTick !== 0 && staleAgeMs > STALE_MS - - const taskCounts = useMemo(() => { - return STATUSES.map((status) => { - const buf = bossSeries.get(seriesKey('agentfm_tasks_total', { status })) - return { status, value: buf ? latestValue(buf) ?? 0 : 0, buf } - }) - }, [bossSeries]) - - const totalTasks = taskCounts.reduce((a, b) => a + b.value, 0) - - const tasksPerMin = useMemo(() => { - const okBuf = bossSeries.get(seriesKey('agentfm_tasks_total', { status: 'ok' })) - return okBuf ? computeTasksPerMinute(okBuf) : 0 - }, [bossSeries]) - - const p95Duration = useMemo(() => { - const buckets: { le: number; count: number }[] = [] - for (const [k, buf] of bossSeries) { - if (!k.startsWith('agentfm_task_duration_seconds_bucket{')) continue - const m = k.match(/le=([^,}]+)/) - if (!m) continue - const leStr = m[1] - const le = leStr === '+Inf' ? Infinity : Number(leStr) - buckets.push({ le, count: latestValue(buf) ?? 0 }) - } - return computeP95FromBuckets(buckets) - }, [bossSeries]) - - const workersOnline = - latestValue(bossSeries.get(seriesKey('agentfm_workers_online', {})) ?? emptyBuf()) ?? 0 - - const streamErrorsTotal = useMemo(() => { - let n = 0 - for (const [k, buf] of bossSeries) { - if (!k.startsWith('agentfm_stream_errors_total{')) continue - n += latestValue(buf) ?? 0 - } - return n - }, [bossSeries]) - - const artifactBytesPerSec = useMemo(() => { - const buf = bossSeries.get(seriesKey('agentfm_artifact_bytes_sent_total', {})) - return buf ? computeRate(buf) : 0 - }, [bossSeries]) - - const cpuPct = useMemo(() => { - const buf = bossSeries.get(seriesKey('process_cpu_seconds_total', {})) - return buf ? computeRate(buf) * 100 : 0 - }, [bossSeries]) - - const rssBytes = - latestValue(bossSeries.get(seriesKey('process_resident_memory_bytes', {})) ?? emptyBuf()) ?? 0 - const goroutines = - latestValue(bossSeries.get(seriesKey('go_goroutines', {})) ?? emptyBuf()) ?? 0 - - const authAttemptsTotal = useMemo(() => { - let n = 0 - for (const [k, buf] of bossSeries) { - if (!k.startsWith('agentfm_auth_attempts_total{')) continue - n += latestValue(buf) ?? 0 - } - return n - }, [bossSeries]) - - const gcPauseP95 = useMemo(() => { - // go_gc_duration_seconds is a summary with quantile labels (0/0.25/0.5/0.75/1). - // 0.75 is the closest exported quantile to p95; report that. - for (const [k, buf] of bossSeries) { - if (!k.startsWith('go_gc_duration_seconds{')) continue - if (!k.includes('quantile=0.75')) continue - return latestValue(buf) ?? 0 - } - return 0 - }, [bossSeries]) - - const errorsByProtocol = useMemo(() => { - const out = new Map() - for (const [k, buf] of bossSeries) { - if (!k.startsWith('agentfm_stream_errors_total{')) continue - const m = k.match(/protocol=([^,}]+)/) - if (!m) continue - const proto = m[1] - out.set(proto, (out.get(proto) ?? 0) + (latestValue(buf) ?? 0)) - } - return Array.from(out.entries()) - }, [bossSeries]) - - const okBuf = bossSeries.get(seriesKey('agentfm_tasks_total', { status: 'ok' })) - const heroValues = okBuf ? ringToArrays(okBuf).v : [] - - return ( -
-
- DASHBOARD - {stale && ( -
- stale {Math.round(staleAgeMs / 1000)}s -
- )} -
- Live -

- TASKS · {Math.round(totalTasks)} total · ▲ {tasksPerMin.toFixed(1)}/min -

- - -
- TASKS · LAST 5 MIN -
-
- {Math.round(totalTasks)} -
-
- {taskCounts.map((t) => ( - - {t.value} {t.status} - - ))} -
- - - -
- - - -
- -
- - - -
- ERRORS BY PROTOCOL -
- {errorsByProtocol.length === 0 ? ( -
(none)
- ) : ( -
- {errorsByProtocol.map(([proto, n]) => ( -
- {proto} - {Math.round(n)} -
- ))} -
- )} - -
- -
- - - - -
-
- ) -} - -function Tile({ label, value, color }: { label: string; value: string; color: string }) { - return ( - -
- {label} -
-
- {value} -
- - ) -} - -function emptyBuf() { - return createRingBuffer() -} -``` - -- [ ] **Step 4: Run — expect pass** - -```bash -npm test -- dashboard -``` - -Expected: 2 passing. - -- [ ] **Step 5: Typecheck** - -```bash -npm run typecheck -``` - -Expected: no errors. If any primitive (e.g. `NeonCard`, `HeroTitle`, `SectionLabel`) has a different import path, fix the imports — these exist in `src/components/primitives/`. - -- [ ] **Step 6: Commit** - -```bash -git add src/routes/Dashboard.tsx tests/unit/dashboard.test.tsx -git commit -m "feat(desktop): Dashboard route with hero tile and metric grid" -``` - ---- - -## Task 13: Wire navigation, App-level history, and PeerView insertion - -**Files:** -- Modify: `src/components/TabStrip.tsx` -- Modify: `src/App.tsx` -- Modify: `src/routes/PeerView.tsx` - -- [ ] **Step 1: Add the Dashboard tab to TabStrip** - -In `src/components/TabStrip.tsx`, replace the `tabs` array (lines 5–12) with: - -```ts -const tabs = [ - { to: '/radar', label: 'Radar' }, - { to: '/dashboard', label: 'Dashboard' }, - { to: '/chat', label: 'Chat' }, - { to: '/activity', label: 'Activity' }, - { to: '/assets', label: 'Assets' }, - { to: '/status', label: 'Status' }, - { to: '/settings', label: 'Settings' }, -] -``` - -- [ ] **Step 2: Register the route and mount useWorkerHistory in App.tsx** - -In `src/App.tsx`, add to the imports near the other route imports: - -```ts -import Dashboard from './routes/Dashboard' -import { useWorkerHistory } from './hooks/useWorkerHistory' -``` - -Inside the `App` function, after the existing `useBackend` and `useEventStream` calls (around line 21), add: - -```ts - useWorkerHistory() -``` - -Inside the `` block, after the `` line, add: - -```tsx - } /> -``` - -- [ ] **Step 3: Insert TelemetryStrip into PeerView** - -In `src/routes/PeerView.tsx`, add the import near the other component imports: - -```ts -import { TelemetryStrip } from '../components/peer/TelemetryStrip'; -``` - -Inside the JSX, between the existing `` block (line 95) and the `` block (line 98), insert: - -```tsx -
- -
-``` - -- [ ] **Step 4: Typecheck and run tests** - -```bash -npm run typecheck && npm test -``` - -Expected: typecheck clean; all unit tests pass. - -- [ ] **Step 5: Smoke-run the app** - -```bash -npm run dev -``` - -Expected: app launches, sidebar shows `Dashboard` between `Radar` and `Chat`. Click it — the page renders without console errors. Open `/peer/` — telemetry strip is visible between SummaryCard and the All/Ratings/Comments tabs. Close the dev process with Ctrl+C. - -- [ ] **Step 6: Commit** - -```bash -git add src/components/TabStrip.tsx src/App.tsx src/routes/PeerView.tsx -git commit -m "feat(desktop): wire Dashboard route, TabStrip entry, and PeerView strip" -``` - ---- - -## Task 14: E2E happy-path test - -**Files:** -- Create: `tests/e2e/dashboard.spec.ts` - -Existing e2e tests in this repo (e.g. `tests/e2e/happy-path.spec.ts`) launch the full Electron app via `_electron`, point it at a real `agentfm` binary, and wait for backend health before asserting on the UI. Mocking `/metrics` would fight that pattern — instead, exercise the dashboard against the real boss `/metrics` endpoint that's already running for the rest of the suite. - -- [ ] **Step 1: Write the test** - -```ts -// tests/e2e/dashboard.spec.ts -import { test, expect, _electron as electron, ElectronApplication, Page } from '@playwright/test' -import path from 'node:path' - -let app: ElectronApplication -let page: Page - -test.beforeAll(async () => { - app = await electron.launch({ - args: ['.'], - env: { - ...process.env, - AGENTFM_BIN: path.resolve( - __dirname, '..', '..', '..', 'agentfm-core', 'agentfm-go', 'agentfm', - ), - }, - cwd: path.resolve(__dirname, '..', '..'), - }) - page = await app.firstWindow() - await page.waitForLoadState('domcontentloaded') - - await page.waitForFunction( - async () => { - const api = (window as unknown as { - api?: { backend: { health: () => Promise<{ ok: boolean }> } } - }).api - if (!api) return false - try { - const r = await api.backend.health() - return r.ok === true - } catch { - return false - } - }, - { timeout: 30_000, polling: 500 }, - ) - - // Dismiss the "New project" wizard if it appears (first-run state). - const wizard = page.locator('h2:has-text("New project")') - if (await wizard.isVisible({ timeout: 3000 }).catch(() => false)) { - await page.locator('input[placeholder*="Team Mesh"]').fill('E2E Dashboard') - await page.locator('button:has-text("Create project")').click() - await wizard.waitFor({ state: 'hidden', timeout: 15_000 }) - } -}) - -test.afterAll(async () => { - await app?.close() -}) - -test('dashboard tab is reachable and renders the TASKS section', async () => { - // Click the Dashboard tab in TabStrip rather than relying on a keyboard - // shortcut (shortcuts are positional and would have shifted). - await page.locator('a[href="#/dashboard"]').click() - - // The route renders the TASKS hero label immediately, even before any - // metric ticks land — proves the route mounted without crashing. - await expect(page.locator('text=/TASKS/i').first()).toBeVisible({ timeout: 10_000 }) - - // After at least one /metrics poll (2s) we expect to see a numeric - // total in the hero. Accept any digit-bearing string in the hero card. - await expect(async () => { - const heroText = await page.locator('text=/TASKS · LAST 5 MIN/i').locator('..').textContent() - expect(heroText).toMatch(/\d+/) - }).toPass({ timeout: 15_000 }) -}) -``` - -- [ ] **Step 2: Build the agentfm binary if missing** - -The e2e launcher expects `agentfm-core/agentfm-go/agentfm` to exist. If it doesn't: - -```bash -cd /Users/saif/Desktop/agentfm-prod/agentfm-core/agentfm-go -go build -o agentfm ./cmd/agentfm -``` - -- [ ] **Step 3: Run e2e** - -```bash -cd /Users/saif/Desktop/agentfm-prod/agentfm-desktop -npm run test:e2e -- dashboard.spec.ts -``` - -Expected: both tests pass. Total wall time ~30 s (Electron cold-start + backend health wait + one poll cycle). - -- [ ] **Step 4: Commit** - -```bash -git add tests/e2e/dashboard.spec.ts -git commit -m "test(desktop): e2e dashboard reaches /dashboard and renders TASKS" -``` - ---- - -## Final verification checklist - -After Task 14 completes, run this once before opening a PR: - -```bash -cd /Users/saif/Desktop/agentfm-prod/agentfm-desktop -npm run typecheck -npm run lint -npm test -npm run test:e2e -``` - -Then perform the manual checks from the spec ("Manual verification (PR checklist)" section): - -1. Start a real boss + worker locally. `npm run dev`. Open `/dashboard` — sparklines tick every 2 s, no console errors. -2. Stop the boss process. Within ~10 s a "stale Ns" badge appears in the dashboard header; the existing backend-down overlay activates. -3. Restart the boss. Polling resumes; badge clears. -4. Navigate to `/peer/`. Telemetry strip populates within 2 s and shows non-zero values for CPU/GPU/RAM/Queue. -5. Background the app window for one minute. Open DevTools → Network. Confirm no `/metrics` requests during the hidden interval. - -If all five pass, the feature is done. diff --git a/docs/superpowers/specs/2026-05-23-metrics-dashboard-design.md b/docs/superpowers/specs/2026-05-23-metrics-dashboard-design.md deleted file mode 100644 index daeb6b4..0000000 --- a/docs/superpowers/specs/2026-05-23-metrics-dashboard-design.md +++ /dev/null @@ -1,240 +0,0 @@ -# Real-time Metrics Dashboard — Desktop App - -**Status:** Draft · 2026-05-23 -**Scope:** `agentfm-desktop` (renderer only). Zero changes to `agentfm-core/agentfm-go`. - -> **Path convention.** All `src/…` paths in this document are relative to the `agentfm-desktop/` repository root (the Electron app). All `internal/…` paths are relative to `agentfm-core/agentfm-go/`. The spec lives in `agentfm-core/` because AgentFM is a single product spread across two sibling repos. - -## Summary - -Add a real-time observability surface to the Electron desktop app, sourced from the boss API's existing `/metrics` endpoint and the existing `/api/workers` telemetry poll. Two surfaces: - -1. A new top-level **`/dashboard`** route showing boss-level signals (task throughput, durations, stream errors, auth attempts, runtime health). -2. A **``** band on each **`/peer/:peerId`** view showing live CPU/GPU/RAM-free/queue sparklines for that worker. - -The boss API already exposes `/metrics` on port 8080 unauthenticated and CORS-free (see `internal/boss/api.go:157`). No backend work is required. - -## Goals - -- A renderer-only feature; no Go or Electron-main changes. -- Live charts that update every 2 seconds and hold the last 5 minutes of history client-side. -- Negligible bundle and memory footprint (single small chart library, ring buffers in renderer memory). -- Match existing UI rhythm — neon palette, mono labels, NeonCard surfaces — so the dashboard reads as a first-class route alongside Radar / Status. - -## Non-goals - -- No on-disk persistence of metric history. Restarting the app starts the buffer fresh. -- No remote scraping. Boss and the desktop run on the same host; we do not attempt to scrape worker `/metrics` endpoints over libp2p. -- No new boss endpoint or new HTTP route. We consume what `/metrics` already exposes. -- No alerting, no thresholds, no notifications. This is a glance-tool, not an oncall pager. -- No multi-window support. The renderer-driven design assumes a single open window today. - -## Data sources - -| Source | Endpoint | Cadence | Used by | -|---|---|---|---| -| Boss Prometheus metrics | `GET http://127.0.0.1:8080/metrics` | 2 s poll while `/dashboard` visible | `Dashboard.tsx` tiles | -| Worker telemetry snapshots | `GET /api/workers` (polled by existing `useWorkers` React Query hook in `src/lib/query.ts`) | 2 s (existing) | `TelemetryStrip.tsx` on `PeerView` | - -### Metrics consumed (from `internal/metrics/metrics.go`) - -| Family | Type | Labels | Used for | -|---|---|---|---| -| `agentfm_tasks_total` | counter | `status ∈ {ok,error,rejected,timeout}` | Hero tile: tasks-per-min, status breakdown | -| `agentfm_task_duration_seconds` | histogram | (buckets 1s–30min) | P95 duration tile (client-side interp) | -| `agentfm_workers_online` | gauge | — | Workers-online tile | -| `agentfm_artifact_bytes_sent_total` | counter | — | Artifact bytes/sec tile (derivative) | -| `agentfm_stream_errors_total` | counter | `protocol`, `reason` | Stream-errors tile + protocol-breakdown bar | -| `agentfm_auth_attempts_total` | counter | `outcome` | Auth-attempts bar | -| `process_cpu_seconds_total` | counter | — | Runtime CPU% (derivative × 100) | -| `process_resident_memory_bytes` | gauge | — | Runtime RSS | -| `go_goroutines` | gauge | — | Runtime goroutines | -| `go_gc_duration_seconds` | summary | — | Runtime GC p95 | - -`agentfm_dht_queries_total` (relay-only) is intentionally **not** consumed — the desktop talks to a boss, not a relay. - -### Worker profile fields consumed (from existing `WorkerProfile` type in `src/types/api.ts`) - -| Field | Used as | -|---|---| -| `cpu_usage_pct` | CPU sparkline (0–100 %) | -| `gpu_usage_pct` | GPU sparkline (0–100 %) | -| `ram_free_gb` | RAM-free sparkline (GB, shown as "RAM FREE") | -| `current_tasks` | Queue sparkline (count) | -| `peer_id`, `last_seen`, `online` | Buffer key / offline detection | - -## Architecture - -Pure renderer-side. The high-level diagram: - -``` -boss API (existing, :8080) renderer (Electron + React) -├─ GET /metrics ────────► useMetricsPoll (2s) ──► promParse ──┐ -└─ GET /api/workers ────────► useWorkerHistory ───────────────┤ - ▼ - useMetricsStore (Zustand, ring buffers) - │ - ┌─────────────────────────────────────┴────────────┐ - ▼ ▼ - Dashboard.tsx TelemetryStrip.tsx - (UPlotChart + tiles) (SparkLine × 4) -``` - -### Component inventory - -**New files (8):** - -1. `src/lib/promParse.ts` — `parseMetrics(text: string): MetricSample[]`. Pure function. Recognises `# HELP`/`# TYPE` (skipped), counter/gauge lines, and histogram triples (`_bucket{le="..."} N`, `_sum`, `_count`). Defensive: each line is wrapped in try/catch; malformed or unknown-name lines are silently dropped. - -2. `src/hooks/useMetricsPoll.ts` — owns the 2 s `setInterval` against `/metrics`. Pauses when `document.visibilityState !== 'visible'`. Backs off to 10 s after 3 consecutive errors; returns to 2 s on first success. - -3. `src/lib/metricsStore.ts` — new Zustand store. Two ring buffers: - - `bossSeries: Map` keyed by `name+labels` (e.g. `agentfm_tasks_total{status=ok}`). Each `RingBuffer = { ts: Float64Array(150), v: Float64Array(150), head: number }` — 150 = 5 min ÷ 2 s. - - `peerSeries: Map>` populated by `useWorkerHistory`. - - `pushBoss(ts, samples)` advances every known series. Series not present in this tick get filled with their previous value carried forward (so charts never gap on a single missing data point). - -4. `src/components/charts/UPlotChart.tsx` — thin React wrapper around `uplot`. Props: `series: RingBuffer[]`, `labels: string[]`, `colors: string[]`, `height`. One `uPlot` instance per chart, kept in a `ref`; `setData()` called via a `useEffect` that subscribes to the relevant store slice. Cleanup calls `chart.destroy()` to survive React 18 StrictMode double-invoke. - -5. `src/components/charts/SparkLine.tsx` — minimal canvas-based single-series sparkline (~30 lines). Used for the 4 PeerView cells where uPlot would be overkill. - -6. `src/routes/Dashboard.tsx` — the Hero+tiles layout: - - Hero card: `agentfm_tasks_total` sum across statuses (5-min sparkline) + status breakdown (ok/error/rejected/timeout). - - Row 2 (3 tiles): P95 task duration (histogram interp), Workers online (`agentfm_workers_online`), Stream errors total (5-min delta). - - Row 3 (3 tiles): Errors by protocol (small stacked bar), Auth attempts (small bar), Artifact bytes/sec (rate, sparkline). - - Row 4 (4 small tiles): CPU%, RSS, Goroutines, GC pause p95. - -7. `src/components/peer/TelemetryStrip.tsx` — 4-cell strip (CPU/GPU/RAM/Queue) using `SparkLine`. Shows the existing "Waiting for telemetry beacon…" placeholder when buffer empty. Shows "(offline — last seen Nm ago)" when `Date.now() - lastTickTs > 30_000`. Re-renders subscribed to one store slice (`peerSeries.get(peerId)`). - -8. `src/hooks/useWorkerHistory.ts` — runs continuously app-wide (mounted in `App.tsx`). Subscribes to the existing React Query `/api/workers` cache; on every successful refetch, iterates the profile list and pushes each peer's CPU/GPU/RAM-free/queue into `peerSeries`. Decoupling history capture from PeerView mount means navigating to a peer page shows charts immediately. - -**Modified files (3):** - -9. `src/components/TabStrip.tsx` — add a `Dashboard` entry to the `tabs` array between Radar and Activity. Match existing label rhythm (no icon column — TabStrip is text-only). - -10. `src/App.tsx` — register `} />`. Call `useWorkerHistory()` directly inside `App` so it runs for the lifetime of the renderer (no wrapper component needed). - -11. `src/routes/PeerView.tsx` — insert `` between the existing `` (line 95) and `` (line 98). - -**New dependency:** - -- `uplot` (~45 KB minified). No `react-uplot`; we wrap directly. - -## Data flow - -### Boss /metrics pipeline (every 2 s, while `/dashboard` is visible) - -``` -fetch('/metrics') - └─→ text/plain - │ - ▼ -parseMetrics(text) - └─→ [{name, labels, value, type}, ...] - │ - ▼ -metricsStore.pushBoss(Date.now(), samples) - └─→ for each series: ring[head]={ts,v}; head=(head+1)%150 - missing series: carry-forward previous value - │ - ▼ -Zustand subscribers re-render - └─→ chart.setData(buffer) - reads buffer.latestValue() -``` - -### Per-worker telemetry pipeline (always running, app-wide) - -``` -React Query /api/workers (existing 2s poll) - └─→ WorkerProfile[] - │ - ▼ -useWorkerHistory pushes every snapshot - └─→ for each profile: peerSeries[peerId][metric].push(now, value) - │ - ▼ - subscribed to peerSeries[x] -``` - -### Derived values (computed at render-time) - -| Derived | Source | Computation | -|---|---|---| -| Tasks-per-minute | `agentfm_tasks_total` counter buffer | `(latest − value_at(now − 60s)) / 60` | -| Task duration p95 | `agentfm_task_duration_seconds_bucket` | Linear interp on cumulative-count vector | -| Artifact bytes/sec | `agentfm_artifact_bytes_sent_total` | First derivative over buffer Δt | -| CPU% | `process_cpu_seconds_total` | First derivative × 100 | - -All four wrapped in `useMemo` keyed to the relevant buffer's `head`. - -### Pause / resume - -- `document.visibilityState === 'hidden'` → polling pauses, no network requests. Store retained. -- Visible again → next tick is a wider Δ; derivatives absorb naturally. -- User navigates away from `/dashboard` → `/metrics` polling stops. Per-worker history keeps going (it piggybacks on `/api/workers` which always polls). - -## Error handling - -| Failure | Behaviour | -|---|---| -| Boss unreachable (network error / non-2xx) | Existing `useBackend` `BackendDownOverlay` covers the whole route. `useMetricsPoll` independently backs off to 10 s after 3 errors; recovers on first success. | -| Single failed poll | Carry-forward keeps charts gap-free. A "stale Ns" badge appears in the dashboard header once last-success > 10 s. | -| Malformed `/metrics` line | `parseMetrics` skips the line silently. In `import.meta.env.DEV`, log a single `console.warn` on first unknown metric name encountered. | -| Unknown metric name (e.g. new metric added Go-side) | Ignored. Adding a chart for it later requires only a config addition; no parser change. | -| Peer disappears from `/api/workers` > 30 s | `TelemetryStrip` shows "(offline — last seen Nm ago)" and stops appending. Buffer retained 5 min so quick disconnect/reconnect preserves history. | -| Peer never published telemetry | Strip shows "Waiting for telemetry beacon…" placeholder (reuses radar-skeleton copy). | -| Equivocator peer | Strip still renders (telemetry is data, not trust). Existing red equivocator banner above provides the warning context. | -| uPlot instance leak under React 18 StrictMode | `useEffect` cleanup calls `chart.destroy()`. | -| `/metrics` returns HTML (proxy mishap, wrong port) | Every line fails to parse; tick produces zero samples; stale badge surfaces. | - -**Explicitly out of scope:** no toasts, no retry-with-backoff for individual fetches (the 2 s loop is its own backoff), no fallback parser. - -## Testing - -Match the existing layout in `tests/{unit,e2e}/`. Test framework: Vitest (unit) + Playwright (e2e), already configured in `package.json`. - -### Unit (Vitest) - -**`promParse.ts`** — the only piece with non-trivial parsing: -- Real `/metrics` output captured from a running boss, saved as `tests/unit/fixtures/metrics-sample.txt`. Round-trip asserts the expected sample count and types. -- Histogram with `_bucket`/`_sum`/`_count` lines collapses into one histogram entity. -- `# HELP` and `# TYPE` lines skipped without error. -- Malformed line mid-file → that line dropped, rest parses. -- Empty input → returns `[]`. - -**`metricsStore.ts` ring buffer:** -- 150 pushes fill the buffer; push #151 overwrites slot 0; `latest()` returns push #151. -- Carry-forward: pushing a tick with a subset of series leaves untouched series with their previous value at the new timestamp. -- Per-peer isolation: pushes to peer A do not appear in peer B's series. - -**Derived computations** (`computeRate`, `computeP95FromBuckets`, `computeTasksPerMinute`) — table-driven tests covering empty buffer, single sample, steady rate. - -### Component smoke (Vitest + Testing Library) - -- `` renders without crashing on empty store (shows skeletons). -- `` renders expected tiles when store is seeded with one tick. -- `` shows placeholder when no buffer; strip when buffer has data; offline copy when `lastTickAgo > 30s`. - -These are smoke tests, not full DOM snapshots. - -### E2E (Playwright) - -One happy-path test: `tests/e2e/dashboard.spec.ts` intercepts `/metrics` via Playwright route-mocking, serves the saved fixture, navigates to `/dashboard`, asserts the hero tile shows the expected task count. Waits one poll cycle (2 s); asserts the chart re-rendered with a second sample. Mocking is preferable to spinning up a real boss for a renderer-only feature. - -### Manual verification (PR checklist) - -- Real worker + boss running locally; open `/dashboard`; watch 30 s — sparklines move, no console errors. -- Stop boss → "stale Ns" badge appears within ~10 s; backend-down overlay activates. -- Restart boss → polling resumes; badge clears. -- Open PeerView for an online worker → telemetry strip populates within 2 s. -- Background the window for a minute → no `/metrics` requests in DevTools network tab. - -### Explicitly out of scope - -- No screenshot / visual-regression tests — uPlot canvas output is noisy under headless rendering. -- No load tests. -- No tests for uPlot itself — trust the library. - -## Open questions - -None at design close — every fork has been resolved during brainstorming. Implementation plan to follow in a separate document.