From 0d09b55a5a621a28b21b7466f4fcda9d975cedca Mon Sep 17 00:00:00 2001 From: dave Date: Wed, 29 Jul 2026 12:27:25 +0100 Subject: [PATCH] feat(orchestrator): add GET /metrics endpoint with task and step counters The orchestrator runs long-lived on Render with no way to observe health beyond tailing logs. Adds a dependency-free in-process counter module and exposes it as JSON at GET /metrics. Counters cover process uptime and memory, task totals (total/active/ completed/failed/interrupted), step outcomes (executed/failed/timed-out), total USDC released, and a step-duration summary with p50/p95/max. Task transitions are keyed on task_id in a Set rather than a bare count, so they are idempotent: a task that fails mid-flight leaves `active` exactly once and can never sit in both `active` and `failed`. Every executor failure path already funnels through makeFailedResult, so that is where failures are counted. Step durations go into a 1024-entry ring buffer, keeping memory bounded while percentiles track recent behaviour. On startup the counters are seeded from the activity log and stored task results so a redeploy does not zero totals that are still knowable; tasks the log shows as in flight are counted as `interrupted` rather than `active`, since nothing is running in a fresh process. --- docs/development.md | 53 ++++ packages/orchestrator/src/executor.ts | 8 + packages/orchestrator/src/metrics.test.ts | 273 ++++++++++++++++++ packages/orchestrator/src/metrics.ts | 259 +++++++++++++++++ .../orchestrator/src/server.metrics.test.ts | 77 +++++ packages/orchestrator/src/server.ts | 32 +- packages/orchestrator/src/task-results.ts | 5 + 7 files changed, 706 insertions(+), 1 deletion(-) create mode 100644 packages/orchestrator/src/metrics.test.ts create mode 100644 packages/orchestrator/src/metrics.ts create mode 100644 packages/orchestrator/src/server.metrics.test.ts diff --git a/docs/development.md b/docs/development.md index d08f5be..9b6923d 100644 --- a/docs/development.md +++ b/docs/development.md @@ -180,6 +180,56 @@ curl -s -X POST http://localhost:3000/api/tasks/preview \ -d '{"prompt": "summarise yesterday Stellar DEX volume", "budget": 1.0}' | jq . ``` +### GET /metrics + +Operational counters for the orchestrator process — task throughput, step +outcomes, USDC released, and step-latency percentiles. Intended for a status +dashboard or an alerting rule; no authentication, no user address required. + +The response is **plain JSON** (not Prometheus text exposition). The shape is +stable — fields may be added, but existing ones keep their names and meaning. + +```json +{ + "uptime_seconds": 3612, + "tasks": { "total": 42, "active": 1, "completed": 38, "failed": 2, "interrupted": 1 }, + "steps": { "executed": 126, "failed": 4, "timed_out": 2 }, + "usdc_released_total": 2.34, + "step_duration_ms": { "count": 126, "p50_ms": 840, "p95_ms": 15000, "max_ms": 21400 }, + "memory": { "rss_bytes": 91234304, "heap_used_bytes": 42118400 } +} +``` + +Field notes: + +| Field | Meaning | +|---|---| +| `tasks.total` | Tasks submitted, including those seeded from the activity log | +| `tasks.active` | In flight **in this process** right now | +| `tasks.interrupted` | Were in flight when a previous process exited — seeded at startup, never incremented at runtime | +| `steps.executed` | Step attempts that finished, successfully or not | +| `steps.failed` | Subset of `executed` that failed | +| `steps.timed_out` | Subset of `failed` whose error text reads as a timeout | +| `usdc_released_total` | USDC released from the vault to the orchestrator wallet | +| `step_duration_ms` | Percentiles over the most recent 1024 step attempts (fixed-size ring, so memory is bounded); `null` until the first step runs | + +Counters are per-process and dependency-free (`packages/orchestrator/src/metrics.ts` +— no metrics library). On startup they are seeded from `data/activity-log.json` +and `data/task-results.json` so a redeploy doesn't zero the totals. Because +activity events are only written for tasks that carry a `user_address`, +anonymous tasks contribute to live counters but are not restored across a +restart. + +`tasks.total` is not guaranteed to equal `active + completed + failed + +interrupted`: a task interrupted by a restart is counted in `total` when it +starts and in `interrupted` only after the *next* startup reads the log. + +**Example curl** + +```bash +curl -s http://localhost:3000/metrics | jq . +``` + ## Testing Unit tests use [Vitest](https://vitest.dev/) and are colocated with the code @@ -193,6 +243,9 @@ to verify in isolation: - `packages/orchestrator/src/validator.test.ts` — execution plan validation. - `packages/orchestrator/src/server.preview.test.ts` — `/api/tasks/preview` endpoint (happy path, no-agents 503, infeasible 422). +- `packages/orchestrator/src/metrics.test.ts` — counter transitions, timeout + classification, percentile math, ring-buffer bounding, and startup seeding. +- `packages/orchestrator/src/server.metrics.test.ts` — `/metrics` response shape. Run the full suite with `npm test`, or scope to a package with `npm test -w packages/registry`. diff --git a/packages/orchestrator/src/executor.ts b/packages/orchestrator/src/executor.ts index 2345803..65231ce 100644 --- a/packages/orchestrator/src/executor.ts +++ b/packages/orchestrator/src/executor.ts @@ -27,6 +27,7 @@ import { makeX402Payment } from './x402-client.js'; import { makeMPPPayment } from './mpp-client.js'; import { rateResponse } from './rater.js'; import { releasePayment, VAULT_ACTIVE } from './agent-vault-client.js'; +import { stepExecuted, stepFailed, usdcReleased } from './metrics.js'; // ── Types ──────────────────────────────────────────────────────────────────── @@ -279,6 +280,7 @@ export class PlanExecutor extends EventEmitter { } releaseHash = typeof released === 'string' ? released : null; + usdcReleased(amountUsdc); // Wrap emit in try/catch — a serialization error must never kill a step try { this.emit('budget_released', { @@ -322,6 +324,7 @@ export class PlanExecutor extends EventEmitter { } const latency_ms = Date.now() - stepStart; + stepExecuted(latency_ms); const quality_rating = await rateResponse(step.action, output); const result: StepResult = { @@ -383,7 +386,12 @@ export class PlanExecutor extends EventEmitter { } } + /** + * Build the StepResult for a failed step. Every failure path routes through + * here, so this is also where the failure is counted. + */ private makeFailedResult(step: ExecutionStep, error: string, latency_ms: number): StepResult { + stepFailed(error, latency_ms); return { step_id: step.step_id, agent_id: step.agent_id, diff --git a/packages/orchestrator/src/metrics.test.ts b/packages/orchestrator/src/metrics.test.ts new file mode 100644 index 0000000..278d03f --- /dev/null +++ b/packages/orchestrator/src/metrics.test.ts @@ -0,0 +1,273 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { + getMetrics, + resetMetrics, + seedMetrics, + stepExecuted, + stepFailed, + taskCompleted, + taskFailed, + taskStarted, + usdcReleased, + isTimeoutError, +} from './metrics.js'; + +beforeEach(() => { + resetMetrics(); +}); + +describe('task counters', () => { + it('starts at zero', () => { + const m = getMetrics(); + expect(m.tasks).toEqual({ total: 0, active: 0, completed: 0, failed: 0, interrupted: 0 }); + expect(m.steps).toEqual({ executed: 0, failed: 0, timed_out: 0 }); + expect(m.usdc_released_total).toBe(0); + }); + + it('moves a task from active to completed', () => { + taskStarted('t1'); + expect(getMetrics().tasks).toMatchObject({ total: 1, active: 1, completed: 0 }); + + taskCompleted('t1'); + expect(getMetrics().tasks).toMatchObject({ total: 1, active: 0, completed: 1, failed: 0 }); + }); + + it('moves a failed task out of active, not into both', () => { + taskStarted('t1'); + taskFailed('t1'); + + const m = getMetrics(); + expect(m.tasks.active).toBe(0); + expect(m.tasks.failed).toBe(1); + expect(m.tasks.completed).toBe(0); + }); + + it('is idempotent on repeated terminal transitions', () => { + taskStarted('t1'); + taskFailed('t1'); + taskFailed('t1'); + taskCompleted('t1'); + + const m = getMetrics(); + expect(m.tasks).toMatchObject({ total: 1, active: 0, failed: 1, completed: 0 }); + }); + + it('ignores a terminal transition for a task that never started', () => { + taskCompleted('never-seen'); + expect(getMetrics().tasks).toMatchObject({ total: 0, completed: 0 }); + }); + + it('does not double-count a repeated start', () => { + taskStarted('t1'); + taskStarted('t1'); + expect(getMetrics().tasks).toMatchObject({ total: 1, active: 1 }); + }); + + it('tracks concurrent tasks independently', () => { + taskStarted('a'); + taskStarted('b'); + taskStarted('c'); + expect(getMetrics().tasks.active).toBe(3); + + taskCompleted('b'); + const m = getMetrics(); + expect(m.tasks).toMatchObject({ total: 3, active: 2, completed: 1 }); + }); + + it('keeps counts correct when terminal transitions interleave out of order', async () => { + taskStarted('a'); + taskStarted('b'); + + await Promise.all([ + (async () => { + await Promise.resolve(); + taskFailed('b'); + })(), + (async () => { + taskCompleted('a'); + })(), + ]); + + expect(getMetrics().tasks).toMatchObject({ total: 2, active: 0, completed: 1, failed: 1 }); + }); +}); + +describe('step counters', () => { + it('counts successes toward executed only', () => { + stepExecuted(100); + stepExecuted(200); + expect(getMetrics().steps).toEqual({ executed: 2, failed: 0, timed_out: 0 }); + }); + + it('counts failures toward executed and failed', () => { + stepExecuted(100); + stepFailed('Agent health check failed: http://x/health', 50); + expect(getMetrics().steps).toEqual({ executed: 2, failed: 1, timed_out: 0 }); + }); + + it('classifies timeout errors as a subset of failures', () => { + stepFailed('The operation was aborted due to timeout', 15000); + stepFailed('Request timed out', 15000); + stepFailed('Agent not found: agent-x', 1); + + expect(getMetrics().steps).toEqual({ executed: 3, failed: 3, timed_out: 2 }); + }); + + it('recognises timeout phrasings without false positives', () => { + expect(isTimeoutError('TimeoutError: signal timed out')).toBe(true); + expect(isTimeoutError('operation was aborted')).toBe(true); + expect(isTimeoutError('timed-out waiting for agent')).toBe(true); + expect(isTimeoutError('Vault release failed for step 2')).toBe(false); + expect(isTimeoutError('')).toBe(false); + expect(isTimeoutError(null)).toBe(false); + }); +}); + +describe('usdc released', () => { + it('accumulates released payments', () => { + usdcReleased(0.02); + usdcReleased(0.03); + usdcReleased(0.02); + expect(getMetrics().usdc_released_total).toBe(0.07); + }); + + it('ignores non-positive and non-finite amounts', () => { + usdcReleased(0); + usdcReleased(-1); + usdcReleased(Number.NaN); + expect(getMetrics().usdc_released_total).toBe(0); + }); +}); + +describe('step duration summary', () => { + it('reports nulls when no steps have run', () => { + expect(getMetrics().step_duration_ms).toEqual({ + count: 0, + p50_ms: null, + p95_ms: null, + max_ms: null, + }); + }); + + it('computes p50, p95 and max over recorded samples', () => { + for (let i = 1; i <= 100; i++) stepExecuted(i); + + const summary = getMetrics().step_duration_ms; + expect(summary.count).toBe(100); + expect(summary.p50_ms).toBe(50); + expect(summary.p95_ms).toBe(95); + expect(summary.max_ms).toBe(100); + }); + + it('includes failed-step durations', () => { + stepExecuted(10); + stepFailed('boom', 90); + expect(getMetrics().step_duration_ms).toMatchObject({ count: 2, max_ms: 90 }); + }); + + it('bounds memory at the ring capacity while executed keeps counting', () => { + for (let i = 0; i < 5000; i++) stepExecuted(i); + + const m = getMetrics(); + expect(m.steps.executed).toBe(5000); + expect(m.step_duration_ms.count).toBe(1024); + // The ring retains the most recent samples, so early values are gone + expect(m.step_duration_ms.max_ms).toBe(4999); + expect(m.step_duration_ms.p50_ms).toBeGreaterThan(4000); + }); + + it('ignores negative durations', () => { + stepExecuted(-5); + expect(getMetrics().step_duration_ms.count).toBe(0); + }); +}); + +describe('uptime and memory', () => { + it('reports a non-negative uptime and real memory readings', () => { + const m = getMetrics(); + expect(m.uptime_seconds).toBeGreaterThanOrEqual(0); + expect(m.memory.rss_bytes).toBeGreaterThan(0); + expect(m.memory.heap_used_bytes).toBeGreaterThan(0); + }); +}); + +describe('seedMetrics', () => { + it('restores task totals and spend from the activity pulse', () => { + seedMetrics({ + pulse: { + total_tasks: 10, + total_completed: 7, + total_failed: 2, + active_tasks: 1, + total_spent_usdc: 0.35, + }, + }); + + const m = getMetrics(); + expect(m.tasks).toEqual({ + total: 10, + active: 0, + completed: 7, + failed: 2, + interrupted: 1, + }); + expect(m.usdc_released_total).toBe(0.35); + }); + + it('counts tasks in flight at shutdown as interrupted, never active', () => { + seedMetrics({ + pulse: { + total_tasks: 3, + total_completed: 1, + total_failed: 0, + active_tasks: 2, + total_spent_usdc: 0, + }, + }); + + expect(getMetrics().tasks.active).toBe(0); + expect(getMetrics().tasks.interrupted).toBe(2); + }); + + it('seeds step counters and durations from stored task results', () => { + seedMetrics({ + taskResults: [ + { + steps: [ + { success: true, error: null, latency_ms: 100 }, + { success: false, error: 'signal timed out', latency_ms: 15000 }, + ], + }, + { + steps: [{ success: false, error: 'Agent not found: x', latency_ms: 5 }], + }, + ], + }); + + const m = getMetrics(); + expect(m.steps).toEqual({ executed: 3, failed: 2, timed_out: 1 }); + expect(m.step_duration_ms).toMatchObject({ count: 3, max_ms: 15000 }); + }); + + it('adds to live counters rather than replacing them', () => { + taskStarted('live'); + taskCompleted('live'); + + seedMetrics({ + pulse: { + total_tasks: 4, + total_completed: 4, + total_failed: 0, + active_tasks: 0, + total_spent_usdc: 0.1, + }, + }); + + expect(getMetrics().tasks).toMatchObject({ total: 5, completed: 5 }); + }); + + it('tolerates an empty seed', () => { + expect(() => seedMetrics({})).not.toThrow(); + expect(getMetrics().tasks.total).toBe(0); + }); +}); diff --git a/packages/orchestrator/src/metrics.ts b/packages/orchestrator/src/metrics.ts new file mode 100644 index 0000000..6c5b1c5 --- /dev/null +++ b/packages/orchestrator/src/metrics.ts @@ -0,0 +1,259 @@ +/** + * In-process metrics counters for the orchestrator. + * + * Dependency-free by design: a plain counters object with increment helpers, + * exposed as JSON by `GET /metrics`. Task lifecycle transitions are driven from + * `server.ts`'s `runTask()`; step and payment counters from `executor.ts`. + * + * Counters are per-process and reset on restart. `seedMetrics()` restores what + * is still knowable from the persisted stores so a redeploy doesn't zero the + * totals. + */ + +/** Number of step-duration samples retained for percentile calculation. */ +const DURATION_SAMPLE_CAPACITY = 1024; + +export interface StepDurationSummary { + /** Samples currently retained (never exceeds DURATION_SAMPLE_CAPACITY). */ + count: number; + p50_ms: number | null; + p95_ms: number | null; + max_ms: number | null; +} + +export interface MetricsSnapshot { + uptime_seconds: number; + tasks: { + /** Tasks submitted (including those seeded from the activity log). */ + total: number; + /** Currently in flight in this process. */ + active: number; + completed: number; + failed: number; + /** Tasks that were in flight when a previous process exited. */ + interrupted: number; + }; + steps: { + /** Step attempts that finished, successfully or not. */ + executed: number; + failed: number; + /** Subset of `failed` whose error looks like a timeout. */ + timed_out: number; + }; + usdc_released_total: number; + step_duration_ms: StepDurationSummary; + memory: { + rss_bytes: number; + heap_used_bytes: number; + }; +} + +interface Counters { + tasks_total: number; + tasks_completed: number; + tasks_failed: number; + tasks_interrupted: number; + steps_executed: number; + steps_failed: number; + steps_timed_out: number; + usdc_released_total: number; +} + +function emptyCounters(): Counters { + return { + tasks_total: 0, + tasks_completed: 0, + tasks_failed: 0, + tasks_interrupted: 0, + steps_executed: 0, + steps_failed: 0, + steps_timed_out: 0, + usdc_released_total: 0, + }; +} + +let counters = emptyCounters(); +let startedAt = Date.now(); + +/** + * Task IDs currently in flight. Tracking IDs rather than a bare count makes the + * terminal transitions idempotent: a task that fails after partially completing + * can only be counted once, and can never sit in `active` and `failed` at the + * same time. + */ +let activeTasks = new Set(); + +/** + * Ring buffer of recent step durations. Bounded memory: once full, the oldest + * sample is overwritten, so percentiles describe the most recent + * DURATION_SAMPLE_CAPACITY steps rather than all history. + */ +let durationSamples: number[] = []; +let durationCursor = 0; + +const TIMEOUT_PATTERN = /\btimed?[\s-]?out\b|\btimeout\b|aborted/i; + +/** True if `error` reads like a timeout rather than a generic failure. */ +export function isTimeoutError(error: string | null | undefined): boolean { + return !!error && TIMEOUT_PATTERN.test(error); +} + +/** Record a task entering the pipeline. */ +export function taskStarted(taskId: string): void { + if (activeTasks.has(taskId)) return; + activeTasks.add(taskId); + counters.tasks_total++; +} + +/** + * Record a task reaching a terminal state. No-op if the task was already + * finalised, so the success path and the error handler can both call it. + */ +export function taskCompleted(taskId: string): void { + if (!activeTasks.delete(taskId)) return; + counters.tasks_completed++; +} + +/** Record a task failing. See {@link taskCompleted} for idempotency notes. */ +export function taskFailed(taskId: string): void { + if (!activeTasks.delete(taskId)) return; + counters.tasks_failed++; +} + +/** Record a completed step attempt and its wall-clock duration. */ +export function stepExecuted(latencyMs: number): void { + counters.steps_executed++; + recordDuration(latencyMs); +} + +/** + * Record a failed step attempt. Failures also count toward `steps_executed`; + * `error` is classified so timeouts can be tracked separately. + */ +export function stepFailed(error: string | null | undefined, latencyMs: number): void { + counters.steps_executed++; + counters.steps_failed++; + if (isTimeoutError(error)) counters.steps_timed_out++; + recordDuration(latencyMs); +} + +/** Record USDC released from the vault to the orchestrator wallet. */ +export function usdcReleased(amountUsdc: number): void { + if (!Number.isFinite(amountUsdc) || amountUsdc <= 0) return; + counters.usdc_released_total += amountUsdc; +} + +function recordDuration(latencyMs: number): void { + if (!Number.isFinite(latencyMs) || latencyMs < 0) return; + if (durationSamples.length < DURATION_SAMPLE_CAPACITY) { + durationSamples.push(latencyMs); + return; + } + durationSamples[durationCursor] = latencyMs; + durationCursor = (durationCursor + 1) % DURATION_SAMPLE_CAPACITY; +} + +/** + * Nearest-rank percentile over the retained samples. `p` is a fraction (0.95 + * for p95). Returns null when no samples have been recorded. + */ +function percentile(sorted: number[], p: number): number | null { + if (sorted.length === 0) return null; + const rank = Math.ceil(p * sorted.length); + return sorted[Math.min(sorted.length - 1, Math.max(0, rank - 1))]; +} + +function summariseDurations(): StepDurationSummary { + const sorted = [...durationSamples].sort((a, b) => a - b); + return { + count: sorted.length, + p50_ms: percentile(sorted, 0.5), + p95_ms: percentile(sorted, 0.95), + max_ms: sorted.length > 0 ? sorted[sorted.length - 1] : null, + }; +} + +/** Snapshot of every counter, for `GET /metrics`. */ +export function getMetrics(): MetricsSnapshot { + const mem = process.memoryUsage(); + return { + uptime_seconds: Math.floor((Date.now() - startedAt) / 1000), + tasks: { + total: counters.tasks_total, + active: activeTasks.size, + completed: counters.tasks_completed, + failed: counters.tasks_failed, + interrupted: counters.tasks_interrupted, + }, + steps: { + executed: counters.steps_executed, + failed: counters.steps_failed, + timed_out: counters.steps_timed_out, + }, + // Rounded to stroop precision — floating-point accumulation of many small + // payments otherwise surfaces as 0.06999999999999999. + usdc_released_total: Math.round(counters.usdc_released_total * 1e7) / 1e7, + step_duration_ms: summariseDurations(), + memory: { + rss_bytes: mem.rss, + heap_used_bytes: mem.heapUsed, + }, + }; +} + +export interface MetricsSeed { + /** Aggregate task/spend history, from `activityStore.getPulse()`. */ + pulse?: { + total_tasks: number; + total_completed: number; + total_failed: number; + active_tasks: number; + total_spent_usdc: number; + }; + /** Persisted task results, from `getAllTaskResults()`. */ + taskResults?: Array<{ + steps: Array<{ success: boolean; error: string | null; latency_ms: number }>; + }>; +} + +/** + * Seed counters from persisted state at startup. + * + * Tasks the activity log still shows as active belong to a process that has + * since exited, so they are counted as `interrupted` rather than `active` — + * nothing is in flight in a freshly started process. Step durations are seeded + * from stored results so percentiles are meaningful before the first new task + * runs. + */ +export function seedMetrics(seed: MetricsSeed): void { + const { pulse, taskResults } = seed; + + if (pulse) { + counters.tasks_total += pulse.total_tasks; + counters.tasks_completed += pulse.total_completed; + counters.tasks_failed += pulse.total_failed; + counters.tasks_interrupted += pulse.active_tasks; + counters.usdc_released_total += pulse.total_spent_usdc; + } + + for (const result of taskResults ?? []) { + for (const step of result.steps ?? []) { + if (step.success) counters.steps_executed++; + else { + counters.steps_executed++; + counters.steps_failed++; + if (isTimeoutError(step.error)) counters.steps_timed_out++; + } + recordDuration(step.latency_ms); + } + } +} + +/** Reset every counter. Test-only. */ +export function resetMetrics(): void { + counters = emptyCounters(); + activeTasks = new Set(); + durationSamples = []; + durationCursor = 0; + startedAt = Date.now(); +} diff --git a/packages/orchestrator/src/server.metrics.test.ts b/packages/orchestrator/src/server.metrics.test.ts new file mode 100644 index 0000000..ac56027 --- /dev/null +++ b/packages/orchestrator/src/server.metrics.test.ts @@ -0,0 +1,77 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import type { Server } from 'http'; +import type { AddressInfo } from 'net'; + +import { app } from './server.js'; +import { resetMetrics, stepExecuted, stepFailed, taskStarted, usdcReleased } from './metrics.js'; + +describe('GET /metrics', () => { + let server: Server; + let baseUrl: string; + + beforeAll(async () => { + await new Promise((resolve) => { + server = app.listen(0, () => { + baseUrl = `http://localhost:${(server.address() as AddressInfo).port}`; + resolve(); + }); + }); + }); + + afterAll(() => { + server.close(); + }); + + beforeEach(() => { + resetMetrics(); + }); + + it('returns JSON with the documented shape', async () => { + const res = await fetch(`${baseUrl}/metrics`); + + expect(res.status).toBe(200); + expect(res.headers.get('content-type')).toMatch(/application\/json/); + + const body = await res.json(); + expect(body).toMatchObject({ + uptime_seconds: expect.any(Number), + tasks: { + total: expect.any(Number), + active: expect.any(Number), + completed: expect.any(Number), + failed: expect.any(Number), + interrupted: expect.any(Number), + }, + steps: { + executed: expect.any(Number), + failed: expect.any(Number), + timed_out: expect.any(Number), + }, + usdc_released_total: expect.any(Number), + step_duration_ms: { count: expect.any(Number) }, + memory: { + rss_bytes: expect.any(Number), + heap_used_bytes: expect.any(Number), + }, + }); + }); + + it('reflects live counter state', async () => { + taskStarted('metrics-endpoint-task'); + stepExecuted(120); + stepFailed('signal timed out', 15000); + usdcReleased(0.02); + + const body = await (await fetch(`${baseUrl}/metrics`)).json(); + + expect(body.tasks).toMatchObject({ total: 1, active: 1 }); + expect(body.steps).toEqual({ executed: 2, failed: 1, timed_out: 1 }); + expect(body.usdc_released_total).toBe(0.02); + expect(body.step_duration_ms).toMatchObject({ count: 2, max_ms: 15000, p50_ms: 120 }); + }); + + it('does not require authentication or a user address', async () => { + const res = await fetch(`${baseUrl}/metrics`); + expect(res.status).toBe(200); + }); +}); diff --git a/packages/orchestrator/src/server.ts b/packages/orchestrator/src/server.ts index d138ff2..5e75c3c 100644 --- a/packages/orchestrator/src/server.ts +++ b/packages/orchestrator/src/server.ts @@ -11,6 +11,7 @@ * POST /api/orchestrators — create a personal orchestrator for a user * POST /api/orchestrators/confirm — submit signed on-chain registration XDR * GET /health — liveness check + * GET /metrics — operational counters (JSON) * WS /ws — real-time task event stream */ import 'dotenv/config'; @@ -48,7 +49,13 @@ import { import * as orchestratorStore from './orchestrator-store.js'; import * as activityStore from './activity-store.js'; import { appendVaultTx, getVaultLedger } from './vault-ledger.js'; -import { saveTaskResult, getTaskResults, deleteTaskResult } from './task-results.js'; +import { + saveTaskResult, + getTaskResults, + deleteTaskResult, + getAllTaskResults, +} from './task-results.js'; +import { getMetrics, seedMetrics, taskStarted, taskCompleted, taskFailed } from './metrics.js'; // ── Config ─────────────────────────────────────────────────────────────────── @@ -314,6 +321,11 @@ app.get('/health', (_req, res) => { res.json({ status: 'ok', agent: 'Orchestrator', address: ORCHESTRATOR_ADDRESS }); }); +// Operational counters — see docs/development.md for the response shape +app.get('/metrics', (_req, res) => { + res.json(getMetrics()); +}); + // List agents from registry app.get('/api/agents', async (_req, res) => { try { @@ -965,6 +977,9 @@ app.post('/api/tasks', async (req, res) => { broadcast('task_accepted', { task_id, task, budget: taskBudget }); runTask(task_id, task, taskBudget, user_address ?? null, webhook_url).catch((err) => { + // Backstop: runTask handles its own errors, so reaching here means its + // handler threw. taskFailed is idempotent, so a double call is harmless. + taskFailed(task_id); console.error('[Orchestrator] Task pipeline error:', err.message); broadcast('task_error', { task_id, task, error: err.message }); }); @@ -1076,6 +1091,8 @@ async function runTask( let vaultTaskId: bigint | null = null; const VAULT_CONTRACT_URL = `https://stellar.expert/explorer/testnet/contract/${process.env.AGENT_VAULT_CONTRACT_ID}`; + taskStarted(task_id); + try { // 1. Fetch available agents let agents: AgentRecord[]; @@ -1353,6 +1370,7 @@ async function runTask( // Trigger webhook on complete/failed execution status if (result.status === 'failed') { + taskFailed(task_id); const failedStep = result.steps.find((s) => !s.success); triggerWebhook({ task_id, @@ -1361,6 +1379,7 @@ async function runTask( completed_at: new Date().toISOString(), }); } else { + taskCompleted(task_id); triggerWebhook({ task_id, status: result.status === 'partial' ? 'partial' : 'completed', @@ -1371,6 +1390,9 @@ async function runTask( }); } } catch (err: any) { + // Move the task out of `active` before anything that could itself throw + taskFailed(task_id); + // Try to complete the vault task even on error to unlock funds if (VAULT_ACTIVE && orchestratorKeypair && vaultTaskId !== null) { vaultCompleteTask(orchestratorKeypair, vaultTaskId).catch(() => {}); @@ -1413,6 +1435,14 @@ server.on('upgrade', (request, socket, head) => { }); if (!process.env.VITEST) { + // Restore what the persisted stores still know so a restart doesn't zero the + // totals. Best-effort: metrics must never block startup. + try { + seedMetrics({ pulse: activityStore.getPulse(), taskResults: getAllTaskResults() }); + } catch (err: any) { + console.warn(`[Orchestrator] Metrics seed skipped: ${err.message}`); + } + server.listen(PORT, () => { console.log(`[Orchestrator] Running on port ${PORT}`); console.log(`[Orchestrator] Wallet: ${ORCHESTRATOR_ADDRESS}`); diff --git a/packages/orchestrator/src/task-results.ts b/packages/orchestrator/src/task-results.ts index 723b867..4a47518 100644 --- a/packages/orchestrator/src/task-results.ts +++ b/packages/orchestrator/src/task-results.ts @@ -84,6 +84,11 @@ export function getTaskResults(userAddress: string, limit = 50): TaskResultEntry .reverse(); } +/** Every persisted result, oldest first. Used to seed metrics at startup. */ +export function getAllTaskResults(): TaskResultEntry[] { + return load(); +} + export function deleteTaskResult(taskId: string): boolean { const store = load(); const before = store.length;