diff --git a/.agentworkforce/agents/factory-feature-guardian/persona.json b/.agentworkforce/agents/factory-feature-guardian/persona.json index d70456d..991a897 100644 --- a/.agentworkforce/agents/factory-feature-guardian/persona.json +++ b/.agentworkforce/agents/factory-feature-guardian/persona.json @@ -3,6 +3,16 @@ "intent": "relay-orchestrator", "tags": ["factory", "verification", "proactive", "health", "release-safety", "durability"], "description": "Reads the validated Factory feature catalog from a scoped repository clone, then cycles through every @agent-relay/factory feature hourly. Each provider-confirmed, idempotent Slack check names the exact source surface, named end-to-end procedure, verification tier, and supported safety/topology boundary; exact revisioned progress persists across the exhaustive manifest and resets only after a complete cycle.", + "skills": [ + { + "id": "factory-feature-verification", + "source": ".agentworkforce/features/verify/procedures.md", + "description": "Verify Factory features against their named end-to-end procedures and supported safety boundaries." + } + ], + "relay": { + "agentName": "factory-feature-guardian" + }, "cloud": true, "harness": "opencode", "model": "deepseek-v4-flash-free", diff --git a/package-lock.json b/package-lock.json index 6646657..a8f8722 100644 --- a/package-lock.json +++ b/package-lock.json @@ -15,7 +15,9 @@ "@agent-relay/integration-prompts": "^10.6.4", "@agent-relay/sdk": "^10.6.4", "@agentworkforce/delivery": "^4.1.23", + "@agentworkforce/persona-kit": "^4.1.34", "@agentworkforce/runtime": "^4.1.23", + "@relaycast/a2a": "^6.2.0", "@relayfile/relay-helpers": "^0.4.6", "@relayfile/sdk": "0.10.34", "@relayflows/core": "^1.0.3", @@ -2654,6 +2656,23 @@ "integrity": "sha512-b1UQwcEZ4yCnMCD8DAL1VlbvBJE9/IX4FTIp7BG1xYpf29SLazLSrqUkj4w7Y5y7cCVP6E5tcqqcI0xemPkHug==", "license": "BSD-3-Clause" }, + "node_modules/@relaycast/a2a": { + "version": "6.2.0", + "resolved": "https://registry.npmjs.org/@relaycast/a2a/-/a2a-6.2.0.tgz", + "integrity": "sha512-gFz6yEHYFjxQ+zCcd8ELc7d4LXL2TI8bTOnJ0tkaYjt4krKn68LtOllc8kjxlUI05v2YXCARbZ022T8Y4yDc2g==", + "dependencies": { + "zod": "^4.3.6" + } + }, + "node_modules/@relaycast/a2a/node_modules/zod": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz", + "integrity": "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/colinhacks" + } + }, "node_modules/@relaycast/sdk": { "version": "1.2.0", "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-1.2.0.tgz", diff --git a/package.json b/package.json index d900199..9677648 100644 --- a/package.json +++ b/package.json @@ -80,13 +80,15 @@ "factory-build": "factory-build/run-factory-build.sh" }, "dependencies": { - "@agentworkforce/delivery": "^4.1.23", - "@agentworkforce/runtime": "^4.1.23", "@agent-relay/cloud": "^10.6.4", "@agent-relay/fleet": "^10.6.4", "@agent-relay/harness-driver": "^10.6.4", "@agent-relay/integration-prompts": "^10.6.4", "@agent-relay/sdk": "^10.6.4", + "@agentworkforce/delivery": "^4.1.23", + "@agentworkforce/persona-kit": "^4.1.34", + "@agentworkforce/runtime": "^4.1.23", + "@relaycast/a2a": "^6.2.0", "@relayfile/relay-helpers": "^0.4.6", "@relayfile/sdk": "0.10.34", "@relayflows/core": "^1.0.3", diff --git a/src/fleet/internal-fleet-client.ts b/src/fleet/internal-fleet-client.ts index 6e3d154..d512405 100644 --- a/src/fleet/internal-fleet-client.ts +++ b/src/fleet/internal-fleet-client.ts @@ -8,11 +8,12 @@ import { dirname, join } from 'node:path' import type { BrokerEvent, ListAgent, SendMessageInput, SpawnPtyInput } from '@agent-relay/harness-driver' import type { PreviewConfig } from '../config/schema' -import type { AgentMessage, AgentPidResolution, Capability, FleetClient, FleetTrackedAgent, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult } from '../ports/fleet' +import type { AgentMessage, AgentPidResolution, Capability, FleetClient, FleetTrackedAgent, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult, TeammateAgent, TeammateQuery } from '../ports/fleet' import type { Logger } from '../ports/system' import { normalizeLogger } from '../logging' import { TailscalePreviewManager, type PreviewManager } from '../node/tailscale-preview' import { resolveRelayWorkspaceKey } from './relay-workspace-key' +import { RelaycastTeammateDirectory, type TeammateDirectory } from './teammates' const requireForResolve = createRequire(import.meta.url) @@ -51,6 +52,11 @@ export interface InternalFleetClientOptions { cwd?: string connectionPath?: string workspaceKey?: string + /** Card-aware directory seam; useful with a local broker plus mock directory. */ + teammateDirectory?: TeammateDirectory + directoryBaseUrl?: string + directoryFetch?: typeof globalThis.fetch + directoryTimeoutMs?: number /** Canonical cloud liveness lookup. Injected by tests; derived from workspaceKey in production. */ listCanonicalOnlineAgentNames?: () => Promise /** Local process-liveness probe used to preserve workers that intentionally run without Relay MCP presence. */ @@ -110,6 +116,7 @@ export class InternalFleetClient implements FleetClient { readonly #cwd?: string readonly #connectionPath?: string readonly #workspaceKey?: string + readonly #teammateDirectory?: TeammateDirectory readonly #listCanonicalOnlineAgentNames?: () => Promise readonly #isProcessAlive: (pid: number) => boolean readonly #now: () => number @@ -148,6 +155,15 @@ export class InternalFleetClient implements FleetClient { this.#cwd = options.cwd this.#connectionPath = options.connectionPath this.#workspaceKey = options.workspaceKey + const directoryToken = resolveRelayWorkspaceKey({ workspaceKey: options.workspaceKey }) + this.#teammateDirectory = options.teammateDirectory ?? (directoryToken + ? new RelaycastTeammateDirectory({ + baseUrl: options.directoryBaseUrl ?? process.env.RELAY_BASE_URL, + token: directoryToken, + fetch: options.directoryFetch, + timeoutMs: options.directoryTimeoutMs, + }) + : undefined) if (options.listCanonicalOnlineAgentNames) { this.#listCanonicalOnlineAgentNames = options.listCanonicalOnlineAgentNames } else if (options.workspaceKey) { @@ -359,6 +375,13 @@ export class InternalFleetClient implements FleetClient { } } + async discoverTeammates(query: TeammateQuery): Promise { + if (!this.#teammateDirectory) { + throw new Error('InternalFleetClient teammate discovery requires a directory or Relay workspace key') + } + return await this.#teammateDirectory.discover(query) + } + async createPreview(input: PreviewStartInput): Promise { assertSelfNode(input.node) if (!this.#previewManager) throw new Error('Tailscale preview provider is not configured') diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index b422ab3..cc535f7 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -2,7 +2,8 @@ import { AgentRelay } from '@agent-relay/sdk' import { resolveRelayAgentToken, resolveRelayWorkspaceKey } from './relay-workspace-key' -import type { AgentLifecycleSignal, AgentMessage, Capability, FleetClient, NodeCapability, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult } from '../ports/fleet' +import type { AgentLifecycleSignal, AgentMessage, Capability, FleetClient, NodeCapability, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult, TeammateAgent, TeammateQuery } from '../ports/fleet' +import { RelaycastTeammateDirectory, type TeammateDirectory } from './teammates' import type { RelayActionInvocation, RelayActionInvocationAck, @@ -45,6 +46,10 @@ export interface RelayFleetClientOptions { lifecycleActionName?: string /** Engine base URL override. Absent means the SDK default (cast.agentrelay.com). */ baseUrl?: string + /** Card-aware directory seam. Defaults to Relaycast GET /v1/a2a/directory. */ + teammateDirectory?: TeammateDirectory + directoryFetch?: typeof globalThis.fetch + directoryTimeoutMs?: number /** Timeout for a spawn/release invocation to reach a terminal ack status. */ spawnAckTimeoutMs?: number pollIntervalMs?: number @@ -106,6 +111,7 @@ export class RelayFleetClient implements FleetClient { // therefore createFleet({ backend: 'relay' })) never throws merely because no // token is configured. #messaging: RelayMessaging | undefined + #teammateDirectory: TeammateDirectory | undefined #messagingReady: Promise | undefined #lifecycleActionReady: Promise | undefined #authenticatedAgentName: string @@ -129,6 +135,7 @@ export class RelayFleetClient implements FleetClient { this.#sleep = options.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms))) this.#log = options.log ?? (() => {}) this.#messaging = options.messaging + this.#teammateDirectory = options.teammateDirectory } /** Agents spawned through this client that have not exited or been released. */ @@ -343,6 +350,11 @@ export class RelayFleetClient implements FleetClient { } } + async discoverTeammates(query: TeammateQuery): Promise { + this.#teammateDirectory ??= this.#createTeammateDirectory() + return await this.#teammateDirectory.discover(query) + } + // `from`/`data` are not representable on the agent-scoped messaging surface: // every send is authored by the factory's own agent identity. async sendMessage(input: SendInput): Promise { @@ -437,6 +449,26 @@ export class RelayFleetClient implements FleetClient { } } + #createTeammateDirectory(): TeammateDirectory { + const env = this.#options.env + const token = resolveRelayWorkspaceKey({ + workspaceKey: this.#options.workspaceKey, + ...(env ? { env, activeWorkspaceKey: () => undefined } : {}), + }) ?? resolveRelayAgentToken({ + agentToken: this.#options.agentToken, + ...(env ? { env } : {}), + }) + if (!token) { + throw new Error('RelayFleetClient teammate discovery requires a workspace key or agent token') + } + return new RelaycastTeammateDirectory({ + baseUrl: this.#options.baseUrl, + token, + fetch: this.#options.directoryFetch, + timeoutMs: this.#options.directoryTimeoutMs, + }) + } + #ensureMessaging(): Promise { if (this.#messaging) return Promise.resolve(this.#messaging) this.#messagingReady ??= this.#bootstrapMessaging().catch((error) => { diff --git a/src/fleet/teammates.test.ts b/src/fleet/teammates.test.ts new file mode 100644 index 0000000..3cdd96e --- /dev/null +++ b/src/fleet/teammates.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it, vi } from 'vitest' + +import { FakeFleetClient } from '../testing/fakes' +import { askTeammate, RelaycastTeammateDirectory } from './teammates' + +const directoryRows = [{ + name: 'infra-agent', + address: 'infra-agent', + skills: [{ id: 'infra-watch', name: 'Infra Watch', tags: ['operations'] }], + tags: ['on-call'], + url: 'https://relay.example/a2a/rpc', + kind: 'native', + status: 'online', +}, { + name: 'review-agent', + address: 'review-agent', + skills: [{ id: 'code-review', name: 'Code Review' }], + tags: ['quality'], + url: 'https://review.example/a2a/rpc', + kind: 'a2a', + status: 'active', +}] + +describe('RelaycastTeammateDirectory', () => { + it('sends skill/tag filters and rejects unfiltered server rows client-side', async () => { + const fetch = vi.fn(async (input) => { + const url = new URL(String(input)) + return new Response(JSON.stringify({ + ok: true, + // Relaycast may match a registered card alias that is not repeated in + // the returned row. Preserve that server match for free-text queries. + data: url.searchParams.has('q') ? [directoryRows[1]] : directoryRows, + }), { status: 200 }) + }) + const directory = new RelaycastTeammateDirectory({ + baseUrl: 'https://relay.example/', + token: 'rk_live_test', + fetch, + }) + + await expect(directory.discover({ skill: 'infra-watch' })).resolves.toEqual([ + expect.objectContaining({ name: 'infra-agent', address: 'infra-agent', kind: 'native' }), + ]) + await expect(directory.discover({ skill: 'missing' })).resolves.toEqual([]) + await expect(directory.discover({ tag: 'quality' })).resolves.toEqual([ + expect.objectContaining({ name: 'review-agent', kind: 'a2a' }), + ]) + await expect(directory.discover({ q: 'private-card-alias' })).resolves.toEqual([ + expect.objectContaining({ name: 'review-agent' }), + ]) + + const firstUrl = fetch.mock.calls[0]?.[0] + expect(String(firstUrl)).toBe('https://relay.example/v1/a2a/directory?skill=infra-watch') + expect(fetch.mock.calls[0]?.[1]).toMatchObject({ + headers: expect.objectContaining({ authorization: 'Bearer rk_live_test' }), + }) + expect(String(fetch.mock.calls[3]?.[0])).toBe('https://relay.example/v1/a2a/directory?q=private-card-alias') + }) +}) + +describe('askTeammate', () => { + it('discovers a target, sends the question, and resolves only its reply', async () => { + const fleet = new FakeFleetClient() + fleet.teammates.push({ + name: 'infra-agent', + address: 'infra-agent', + skills: [{ id: 'infra-watch', name: 'Infra Watch' }], + tags: [], + url: 'https://relay.example/a2a/rpc', + kind: 'native', + }) + + const asked = askTeammate(fleet, { + from: 'factory-worker', + question: 'Is the deploy healthy?', + skill: 'infra-watch', + timeoutMs: 1_000, + }) + await vi.waitFor(() => expect(fleet.messages).toHaveLength(1)) + fleet.emitAgentMessage({ from: 'someone-else', target: 'factory-worker', body: 'wrong' }) + fleet.emitAgentMessage({ from: 'infra-agent', target: 'different-worker', body: 'also wrong' }) + fleet.emitAgentMessage({ from: 'infra-agent', target: 'factory-worker', body: 'All systems green.' }) + + await expect(asked).resolves.toMatchObject({ + teammate: { name: 'infra-agent' }, + reply: { from: 'infra-agent', body: 'All systems green.' }, + }) + expect(fleet.messages[0]).toMatchObject({ + to: 'infra-agent', + from: 'factory-worker', + text: 'Is the deploy healthy?', + mode: 'wait', + data: { factoryCapability: 'ask-a-teammate', requester: 'factory-worker' }, + }) + }) + + it('returns a bounded timeout when the teammate never replies', async () => { + vi.useFakeTimers() + try { + const fleet = new FakeFleetClient() + fleet.teammates.push({ + name: 'silent-agent', + address: 'silent-agent', + skills: [{ id: 'infra-watch', name: 'Infra Watch' }], + tags: [], + url: 'https://relay.example/a2a/rpc', + kind: 'native', + }) + const asked = askTeammate(fleet, { + from: 'factory-worker', + question: 'Hello?', + skill: 'infra-watch', + timeoutMs: 50, + }) + const rejected = expect(asked).rejects.toThrow('Timed out waiting for a reply from a teammate after 50ms') + await vi.advanceTimersByTimeAsync(50) + await rejected + } finally { + vi.useRealTimers() + } + }) + + it('refuses to choose an arbitrary teammate without a target or discovery query', async () => { + const fleet = new FakeFleetClient() + await expect(askTeammate(fleet, { + from: 'factory-worker', + question: 'Whoever is first, please answer.', + })).rejects.toThrow('requires a discovered teammate or a skill/tag/query') + }) +}) diff --git a/src/fleet/teammates.ts b/src/fleet/teammates.ts new file mode 100644 index 0000000..db8c51d --- /dev/null +++ b/src/fleet/teammates.ts @@ -0,0 +1,300 @@ +import { randomUUID } from 'node:crypto' + +import { A2aSkillSchema } from '@relaycast/a2a' + +import type { AgentMessage, FleetClient, TeammateAgent, TeammateQuery } from '../ports/fleet' + +export const DEFAULT_RELAYCAST_BASE_URL = 'https://cast.agentrelay.com' +export const DEFAULT_TEAMMATE_DIRECTORY_TIMEOUT_MS = 10_000 +export const DEFAULT_ASK_TEAMMATE_TIMEOUT_MS = 30_000 + +export interface TeammateDirectory { + discover(query: TeammateQuery): Promise +} + +export interface RelaycastTeammateDirectoryOptions { + baseUrl?: string + token?: string + fetch?: typeof globalThis.fetch + timeoutMs?: number +} + +/** Relaycast-backed, card-aware teammate discovery. */ +export class RelaycastTeammateDirectory implements TeammateDirectory { + readonly #baseUrl: string + readonly #token?: string + readonly #fetch: typeof globalThis.fetch + readonly #timeoutMs: number + + constructor(options: RelaycastTeammateDirectoryOptions = {}) { + this.#baseUrl = normalizeBaseUrl(options.baseUrl ?? DEFAULT_RELAYCAST_BASE_URL) + this.#token = nonEmpty(options.token) + this.#fetch = options.fetch ?? globalThis.fetch + this.#timeoutMs = positiveTimeout(options.timeoutMs, DEFAULT_TEAMMATE_DIRECTORY_TIMEOUT_MS) + } + + async discover(query: TeammateQuery): Promise { + const normalized = normalizeQuery(query) + const url = new URL('/v1/a2a/directory', `${this.#baseUrl}/`) + if (normalized.skill) url.searchParams.set('skill', normalized.skill) + if (normalized.tag) url.searchParams.set('tag', normalized.tag) + if (normalized.q) url.searchParams.set('q', normalized.q) + + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), this.#timeoutMs) + try { + const response = await this.#fetch(url, { + method: 'GET', + headers: { + accept: 'application/json', + ...(this.#token ? { authorization: `Bearer ${this.#token}` } : {}), + }, + signal: controller.signal, + }) + const payload = await readJson(response) + if (!response.ok) { + throw new Error(`Relaycast teammate directory returned ${response.status}: ${errorDetail(payload)}`) + } + + const rows = directoryRows(payload) + .map(parseDirectoryEntry) + .filter((entry): entry is TeammateAgent => Boolean(entry)) + // Keep the client honest even if a server version ignores a filter. This + // also makes an unknown exact skill/tag deterministically return []. + .filter((entry) => matchesQuery(entry, normalized)) + + const unique = new Map() + for (const entry of rows) unique.set(`${entry.kind}:${entry.address}`, entry) + return [...unique.values()] + } catch (error) { + if (controller.signal.aborted) { + throw new Error(`Timed out querying the Relaycast teammate directory after ${this.#timeoutMs}ms`, { cause: error }) + } + throw error + } finally { + clearTimeout(timeout) + } + } +} + +export interface AskTeammateInput { + /** Worker identity that is asking the question. */ + from: string + question: string + /** Use an already-discovered target, or provide a query to resolve one. */ + teammate?: TeammateAgent + skill?: string + tag?: string + q?: string + timeoutMs?: number +} + +export interface AskTeammateResult { + requestId: string + teammate: TeammateAgent + reply: AgentMessage +} + +/** + * Resolve a teammate (when needed), deliver a relay DM, and wait for that + * teammate's reply. The listener is armed before sending so fast replies cannot + * race the waiter; the whole operation has one bounded deadline. + */ +export async function askTeammate(fleet: FleetClient, input: AskTeammateInput): Promise { + const timeoutMs = positiveTimeout(input.timeoutMs, DEFAULT_ASK_TEAMMATE_TIMEOUT_MS) + const requestId = randomUUID() + const from = requiredText(input.from, 'askTeammate.from') + const question = requiredText(input.question, 'askTeammate.question') + const query = normalizeQuery({ skill: input.skill, tag: input.tag, q: input.q }) + if (!input.teammate && !query.skill && !query.tag && !query.q) { + throw new Error('askTeammate requires a discovered teammate or a skill/tag/query') + } + const deadline = Date.now() + timeoutMs + + return await new Promise((resolve, reject) => { + let settled = false + let unsubscribe = () => {} + const finish = (result: AskTeammateResult) => { + if (settled) return + settled = true + clearTimeout(timeout) + unsubscribe() + resolve(result) + } + const fail = (error: unknown) => { + if (settled) return + settled = true + clearTimeout(timeout) + unsubscribe() + reject(error) + } + const timeout = setTimeout(() => { + fail(new Error(`Timed out waiting for a reply from a teammate after ${timeoutMs}ms`)) + }, timeoutMs) + + void (async () => { + const teammate = input.teammate ?? (await fleet.discoverTeammates(query))[0] + if (settled) return + if (!teammate) { + throw new Error('No teammate matched the requested skill/tag/query') + } + if (!fleet.onAgentMessage) { + throw new Error('This fleet backend cannot observe teammate replies') + } + unsubscribe = fleet.onAgentMessage((message) => { + if (!sameAgent(message.from, teammate.address) && !sameAgent(message.from, teammate.name)) return + if (!sameAgent(message.target, from)) return + finish({ requestId, teammate, reply: message }) + }) + const send = { + to: teammate.address, + from, + text: question, + mode: 'wait' as const, + data: { + factoryCapability: 'ask-a-teammate', + requestId, + requester: from, + }, + } + if (fleet.waitForInjected) { + await fleet.waitForInjected(send, { timeoutMs: Math.max(1, deadline - Date.now()) }) + } else { + await fleet.sendMessage(send) + } + })().catch(fail) + }) +} + +function normalizeQuery(query: TeammateQuery): TeammateQuery { + return { + skill: nonEmpty(query.skill), + tag: nonEmpty(query.tag), + q: nonEmpty(query.q), + } +} + +function matchesQuery(entry: TeammateAgent, query: TeammateQuery): boolean { + const skill = normalizedComparable(query.skill) + if (skill && !entry.skills.some((candidate) => + normalizedComparable(candidate.id) === skill || normalizedComparable(candidate.name) === skill)) return false + + const tag = normalizedComparable(query.tag) + if (tag) { + const tags = [...entry.tags, ...entry.skills.flatMap((candidate) => candidate.tags ?? [])] + if (!tags.some((candidate) => normalizedComparable(candidate) === tag)) return false + } + + // Relaycast also searches aliases that are intentionally not exposed in a + // directory row (for example, an A2A card name behind its derived ext-* relay + // identity). Re-applying q here would discard those valid server matches. + // Exact skill/tag filters are fully represented by the row, so those remain + // enforced client-side to protect callers from an older unfiltered server. + return true +} + +function parseDirectoryEntry(value: unknown): TeammateAgent | undefined { + const record = asRecord(value) + const name = readString(record, 'name') + const url = readString(record, 'url', 'endpoint_url', 'endpointUrl') + const kind = readString(record, 'kind') + if (!name || !url || (kind !== 'native' && kind !== 'a2a')) return undefined + try { + new URL(url) + } catch { + return undefined + } + + const rawSkills = Array.isArray(record?.skills) ? record.skills : [] + const skills = rawSkills.flatMap((value) => { + const candidate = typeof value === 'string' + ? { id: value, name: value } + : value + const parsed = A2aSkillSchema.safeParse(candidate) + return parsed.success ? [parsed.data] : [] + }) + const tags = readStrings(record, 'tags') + const address = readString(record, 'address', 'relay_name', 'relayName', 'target') ?? name + return { + name, + ...(readString(record, 'description') ? { description: readString(record, 'description') } : {}), + skills, + url, + kind, + address, + tags, + ...(readString(record, 'status') ? { status: readString(record, 'status') } : {}), + ...(readString(record, 'certification') ? { certification: readString(record, 'certification') } : {}), + } +} + +function directoryRows(payload: unknown): unknown[] { + if (Array.isArray(payload)) return payload + const record = asRecord(payload) + return Array.isArray(record?.data) ? record.data : [] +} + +async function readJson(response: Response): Promise { + const text = await response.text() + if (!text) return undefined + try { + return JSON.parse(text) as unknown + } catch (error) { + throw new Error('Relaycast teammate directory returned invalid JSON', { cause: error }) + } +} + +function errorDetail(payload: unknown): string { + const record = asRecord(payload) + const error = asRecord(record?.error) + return readString(error, 'message') ?? readString(record, 'message') ?? 'request failed' +} + +function positiveTimeout(value: number | undefined, fallback: number): number { + return Number.isFinite(value) && (value ?? 0) > 0 ? Math.floor(value as number) : fallback +} + +function normalizeBaseUrl(value: string): string { + return value.replace(/\/+$/u, '') +} + +function sameAgent(left: string, right: string): boolean { + return left.replace(/^@/u, '').toLowerCase() === right.replace(/^@/u, '').toLowerCase() +} + +function normalizedComparable(value: string | undefined): string { + return value?.trim().toLowerCase() ?? '' +} + +function nonEmpty(value: string | undefined): string | undefined { + const trimmed = value?.trim() + return trimmed || undefined +} + +function requiredText(value: string, label: string): string { + const normalized = nonEmpty(value) + if (!normalized) throw new Error(`${label} must be a non-empty string`) + return normalized +} + +function asRecord(value: unknown): Record | undefined { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined +} + +function readString(record: Record | undefined, ...keys: string[]): string | undefined { + if (!record) return undefined + for (const key of keys) { + const value = record[key] + if (typeof value === 'string' && value.trim()) return value.trim() + } + return undefined +} + +function readStrings(record: Record | undefined, key: string): string[] { + const value = record?.[key] + return Array.isArray(value) + ? [...new Set(value.filter((entry): entry is string => typeof entry === 'string' && Boolean(entry.trim())).map((entry) => entry.trim()))] + : [] +} diff --git a/src/index.ts b/src/index.ts index 5fd4cb3..3bc1f9e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -73,6 +73,19 @@ export type { InternalFleetClientOptions, } from './fleet/internal-fleet-client' export { RelayFleetClient } from './fleet/relay-fleet-client' +export { + askTeammate, + DEFAULT_ASK_TEAMMATE_TIMEOUT_MS, + DEFAULT_RELAYCAST_BASE_URL, + DEFAULT_TEAMMATE_DIRECTORY_TIMEOUT_MS, + RelaycastTeammateDirectory, +} from './fleet/teammates' +export type { + AskTeammateInput, + AskTeammateResult, + RelaycastTeammateDirectoryOptions, + TeammateDirectory, +} from './fleet/teammates' export { RelayfileCloudMountClient, resolveFactoryWorkspace, @@ -197,6 +210,11 @@ export { resolveFactoryNodeConfigPath, runRelayflowsWorkflow, } from './node/factory-node' +export { + deriveFactoryPersonaCard, + RelaycastAgentCardPublisher, +} from './node/factory-persona-card' +export { startFactoryNode } from './node/factory-node-runtime' export { TailscalePreviewManager } from './node/tailscale-preview' export type { PreviewCommandRunner, @@ -206,12 +224,23 @@ export type { } from './node/tailscale-preview' export type { FactoryNodeDefinitionOptions, + FactoryNodeDefinition, FactoryNodeInventoryAgent, FactoryNodeInventorySync, WorkflowRunner, WorkflowRunnerInput, WorkflowRunnerResult, } from './node/factory-node' +export type { + AgentCardPublisher, + FactoryPersonaCardInput, + PublishedAgentCard, + RelaycastAgentCardPublisherOptions, +} from './node/factory-persona-card' +export type { + RunningFactoryNode, + StartFactoryNodeOptions, +} from './node/factory-node-runtime' export { assertInFactoryScope, factoryScopeSafety, @@ -278,6 +307,7 @@ export type { ResourceSubscriptionsSdkClientOptions, } from './subscriptions' export type { + A2aSkill, Capability, Environment, EnvironmentProvider, @@ -311,6 +341,8 @@ export type { SendInput, SpawnInput, SpawnResult, + TeammateAgent, + TeammateQuery, GithubRead, GithubIssueStatus, GithubWriteback, diff --git a/src/node/factory-node-runtime.ts b/src/node/factory-node-runtime.ts new file mode 100644 index 0000000..148c363 --- /dev/null +++ b/src/node/factory-node-runtime.ts @@ -0,0 +1,64 @@ +import { startServeNode } from '@agent-relay/fleet' + +import type { FleetNodeInfo, RunningNode, ServeNodeOptions } from '@agent-relay/fleet' + +import type { AgentCardPublisher, PublishedAgentCard } from './factory-persona-card' +import type { FactoryNodeDefinition } from './factory-node' + +export interface StartFactoryNodeOptions extends Omit { + definition: FactoryNodeDefinition + cardPublisher?: AgentCardPublisher + onRegistered?: (info: FleetNodeInfo) => void + /** Test seam; production uses @agent-relay/fleet startServeNode. */ + serve?: (options: ServeNodeOptions) => RunningNode +} + +export interface RunningFactoryNode extends RunningNode { + /** Settles only after the online registration hook publishes the persona card. */ + cardPublished: Promise +} + +/** Start a Factory node and publish its persona card on the node-online edge. */ +export function startFactoryNode(options: StartFactoryNodeOptions): RunningFactoryNode { + let resolvePublication!: (value: PublishedAgentCard | undefined) => void + let rejectPublication!: (reason: unknown) => void + const cardPublished = new Promise((resolve, reject) => { + resolvePublication = resolve + rejectPublication = reject + }) + let registered = false + const serve = options.serve ?? startServeNode + const running = serve({ + definition: options.definition, + connection: options.connection, + ...(options.providerName ? { providerName: options.providerName } : {}), + ...(options.triggers ? { triggers: options.triggers } : {}), + ...(options.nameOverride ? { nameOverride: options.nameOverride } : {}), + ...(options.maxAgentsOverride !== undefined ? { maxAgentsOverride: options.maxAgentsOverride } : {}), + ...(options.reconnect !== undefined ? { reconnect: options.reconnect } : {}), + ...(options.signal ? { signal: options.signal } : {}), + ...(options.logger ? { logger: options.logger } : {}), + ...(options.log ? { log: options.log } : {}), + ...(options.warn ? { warn: options.warn } : {}), + onRegistered(info) { + options.onRegistered?.(info) + if (registered) return + registered = true + const card = options.definition.agentCard + if (!card) { + resolvePublication(undefined) + return + } + if (!options.cardPublisher) { + rejectPublication(new Error('Factory persona node came online without an AgentCardPublisher')) + return + } + void options.cardPublisher.publishAgentCard(card).then(resolvePublication, rejectPublication) + }, + }) + return { + stop: () => running.stop(), + done: running.done, + cardPublished, + } +} diff --git a/src/node/factory-node.ts b/src/node/factory-node.ts index 46cb078..9cac7ce 100644 --- a/src/node/factory-node.ts +++ b/src/node/factory-node.ts @@ -7,6 +7,7 @@ import type { AgentSpec, JsonValue, RestartPolicy, SpawnMode } from '@agent-rela import type { SpawnPtyInput } from '@agent-relay/harness-driver' import { runScriptWorkflow, runWorkflow } from '@relayflows/core' import type { WorkflowRunRow } from '@relayflows/core' +import type { A2aAgentCard } from '@relaycast/a2a' import { z } from 'zod' import { loadFactoryConfig, NodeConfigSchema, type NodeConfig } from '../config/schema' @@ -18,6 +19,7 @@ import { } from '../fleet/internal-fleet-client' import type { Capability, PreviewReference } from '../ports/fleet' import { TailscalePreviewManager, type PreviewManager } from './tailscale-preview' +import { deriveFactoryPersonaCard, type FactoryPersonaCardInput } from './factory-persona-card' export const FACTORY_NODE_CONFIG_ENV = 'FACTORY_NODE_CONFIG' export const AGENT_RELAY_FACTORY_NODE_CONFIG_ENV = 'AGENT_RELAY_FACTORY_NODE_CONFIG' @@ -159,6 +161,12 @@ export interface FactoryNodeDefinitionOptions { workflowRunner?: WorkflowRunner resolveAgentRelayMcpCommand?: () => AgentRelayMcpCommand | undefined previewManager?: PreviewManager + /** Persona hosted by this node; its canonical A2A card is attached for online publication. */ + persona?: FactoryPersonaCardInput +} + +export interface FactoryNodeDefinition extends FleetNodeDefinition { + readonly agentCard?: A2aAgentCard } export interface WorkflowRunnerInput { @@ -193,7 +201,7 @@ export interface FactoryNodeInventorySync { agents: FactoryNodeInventoryAgent[] } -export function createFactoryNodeDefinition(options: FactoryNodeDefinitionOptions): FleetNodeDefinition { +export function createFactoryNodeDefinition(options: FactoryNodeDefinitionOptions): FactoryNodeDefinition { const config = options.config const capabilities = normalizeCapabilities([ ...config.capabilities, @@ -236,13 +244,16 @@ export function createFactoryNodeDefinition(options: FactoryNodeDefinitionOption })) as unknown as FleetCapabilityValue } - return defineNode({ + const definition = defineNode({ name: options.name ?? defaultFactoryNodeName(), ...(options.maxAgents !== undefined ? { maxAgents: options.maxAgents } : {}), capabilities: handlers, tags: options.tags ?? defaultFactoryNodeTags(config), version: options.version ?? 'factory-node-v1', }) + return options.persona + ? { ...definition, agentCard: deriveFactoryPersonaCard(options.persona) } + : definition } async function runPreviewCapability( diff --git a/src/node/factory-persona-card.test.ts b/src/node/factory-persona-card.test.ts new file mode 100644 index 0000000..2048259 --- /dev/null +++ b/src/node/factory-persona-card.test.ts @@ -0,0 +1,156 @@ +import { readFileSync } from 'node:fs' + +import { A2aAgentCardSchema } from '@relaycast/a2a' +import { describe, expect, it, vi } from 'vitest' + +import { createFactoryNodeDefinition, parseFactoryNodeConfig } from './factory-node' +import { deriveFactoryPersonaCard, RelaycastAgentCardPublisher } from './factory-persona-card' +import { startFactoryNode } from './factory-node-runtime' + +const persona = JSON.parse(readFileSync( + new URL('../../.agentworkforce/agents/factory-feature-guardian/persona.json', import.meta.url), + 'utf8', +)) as unknown + +describe('Factory persona cards', () => { + it('matches the canonical mapper for runtime flags and an intent fallback skill', () => { + const cardWithRuntimeFlags = deriveFactoryPersonaCard({ + persona: { + ...(persona as Record), + skills: [], + capabilities: { + streaming: true, + pushNotifications: { enabled: true }, + }, + }, + baseUrl: 'https://agent.example', + version: '1.2.3', + }) + + expect(cardWithRuntimeFlags).toMatchObject({ + skills: [{ + id: 'relay-orchestrator', + name: 'Relay Orchestrator', + }], + capabilities: { + streaming: true, + pushNotifications: true, + }, + }) + + const cardWithoutDeclaredSkills = deriveFactoryPersonaCard({ + persona: { + ...(persona as Record), + skills: [], + capabilities: {}, + }, + baseUrl: 'https://agent.example', + version: '1.2.3', + }) + expect(cardWithoutDeclaredSkills.skills).toEqual([expect.objectContaining({ + id: 'relay-orchestrator', + name: 'Relay Orchestrator', + })]) + }) + + it('derives a shared-schema card and publishes it on the node-online edge', async () => { + const published: unknown[] = [] + const fetch = vi.fn(async (_url, init) => { + published.push(JSON.parse(String(init?.body))) + return new Response(JSON.stringify({ + ok: true, + data: { relay_name: 'ext-factory-feature-guardian-a1b2c3d4', certification: 'level_1' }, + }), { status: 201 }) + }) + const definition = createFactoryNodeDefinition({ + config: parseFactoryNodeConfig({ + workspaceId: 'workspace-1', + capabilities: ['spawn:codex'], + clonePaths: { 'AgentWorkforce/factory': '/work/factory' }, + dryRun: false, + }), + name: 'factory-persona-node', + persona: { + persona, + baseUrl: 'https://relay.example', + version: '1.0.0', + }, + }) + expect(() => A2aAgentCardSchema.parse(definition.agentCard)).not.toThrow() + expect(definition.agentCard).toMatchObject({ + name: 'factory-feature-guardian', + skills: [expect.objectContaining({ id: 'factory-feature-verification' })], + }) + + const publisher = new RelaycastAgentCardPublisher({ + baseUrl: 'https://relay.example', + token: 'rk_live_test', + fetch, + }) + const running = startFactoryNode({ + definition, + connection: { nodeId: 'node-1', nodeToken: 'nt_live_test' }, + cardPublisher: publisher, + serve(options) { + queueMicrotask(() => options.onRegistered?.({ + name: definition.name, + capabilities: Object.keys(definition.capabilities), + })) + return { stop: async () => {}, done: Promise.resolve() } + }, + }) + + await expect(running.cardPublished).resolves.toEqual({ + name: 'factory-feature-guardian', + address: 'ext-factory-feature-guardian-a1b2c3d4', + certification: 'level_1', + }) + expect(published).toEqual([{ agent_card: definition.agentCard }]) + }) + + it('verifies an idempotent registration conflict and returns the existing relay address', async () => { + const definition = createFactoryNodeDefinition({ + config: parseFactoryNodeConfig({ + workspaceId: 'workspace-1', + capabilities: ['spawn:codex'], + clonePaths: { 'AgentWorkforce/factory': '/work/factory' }, + dryRun: false, + }), + persona: { + persona, + baseUrl: 'https://agent.example', + version: '1.0.0', + }, + }) + const fetch = vi.fn(async (_url, init) => { + if (init?.method === 'POST') { + return new Response(JSON.stringify({ + ok: false, + error: { code: 'a2a_agent_already_exists', message: 'already exists' }, + }), { status: 409 }) + } + return new Response(JSON.stringify({ + ok: true, + data: [{ + relay_name: 'ext-factory-feature-guardian-a1b2c3d4', + agent_card: definition.agentCard, + }], + }), { status: 200 }) + }) + const publisher = new RelaycastAgentCardPublisher({ + baseUrl: 'https://relay.example', + token: 'rk_live_test', + fetch, + }) + + await expect(publisher.publishAgentCard(definition.agentCard!)).resolves.toEqual({ + name: 'factory-feature-guardian', + address: 'ext-factory-feature-guardian-a1b2c3d4', + alreadyPublished: true, + }) + expect(fetch.mock.calls.map(([url]) => String(url))).toEqual([ + 'https://relay.example/v1/a2a/register', + 'https://relay.example/v1/a2a/agents', + ]) + }) +}) diff --git a/src/node/factory-persona-card.ts b/src/node/factory-persona-card.ts new file mode 100644 index 0000000..4bbb6bc --- /dev/null +++ b/src/node/factory-persona-card.ts @@ -0,0 +1,298 @@ +import { isDeepStrictEqual } from 'node:util' + +import { A2aAgentCardSchema, type A2aAgentCard } from '@relaycast/a2a' +import * as personaKitSpec from '@agentworkforce/persona-kit/spec' + +import type { PersonaIntent, PersonaSpec } from '@agentworkforce/persona-kit/spec' + +export interface FactoryPersonaCardInput { + /** Raw persona.json data or an already parsed PersonaSpec. */ + persona: unknown + /** Deployed agent origin; A2A RPC is available at `/a2a/rpc`. */ + baseUrl: string + version: string + documentationUrl?: string + inputModes?: string[] + outputModes?: string[] +} + +export interface AgentCardPublisher { + publishAgentCard(card: A2aAgentCard): Promise +} + +export interface PublishedAgentCard { + name: string + address: string + alreadyPublished?: boolean + certification?: string +} + +export interface RelaycastAgentCardPublisherOptions { + baseUrl: string + token: string + fetch?: typeof globalThis.fetch + timeoutMs?: number +} + +type DeriveAgentCardOptions = { + baseUrl: string + version: string + documentationUrl?: string + inputModes?: string[] + outputModes?: string[] +} +type DeriveAgentCard = (persona: PersonaSpec, options: DeriveAgentCardOptions) => A2aAgentCard +const A2A_TRANSPORT_CAPABILITIES = new Set(['streaming', 'pushNotifications']) + +/** + * Derive and schema-validate the card attached to a Factory-hosted persona. + * + * workforce#296 owns the canonical mapper. persona-kit 4.1.34 predates that + * export, so this release keeps a compatibility mapper behind the same call + * boundary. As soon as a persona-kit containing deriveAgentCard is installed, + * it is selected automatically and every result is still parsed by the shared + * @relaycast/a2a schema. + */ +export function deriveFactoryPersonaCard(input: FactoryPersonaCardInput): A2aAgentCard { + const persona = parsePersona(input.persona) + const options: DeriveAgentCardOptions = { + baseUrl: input.baseUrl, + version: input.version, + ...(input.documentationUrl ? { documentationUrl: input.documentationUrl } : {}), + ...(input.inputModes ? { inputModes: input.inputModes } : {}), + ...(input.outputModes ? { outputModes: input.outputModes } : {}), + } + const upstream = (personaKitSpec as typeof personaKitSpec & { deriveAgentCard?: DeriveAgentCard }).deriveAgentCard + return A2aAgentCardSchema.parse( + upstream ? upstream(persona, options) : deriveAgentCardCompatibility(persona, options), + ) +} + +/** Publish an inline canonical card through Relaycast's A2A registration seam. */ +export class RelaycastAgentCardPublisher implements AgentCardPublisher { + readonly #baseUrl: string + readonly #token: string + readonly #fetch: typeof globalThis.fetch + readonly #timeoutMs: number + + constructor(options: RelaycastAgentCardPublisherOptions) { + this.#baseUrl = options.baseUrl.replace(/\/+$/u, '') + this.#token = requiredText(options.token, 'RelaycastAgentCardPublisher token') + this.#fetch = options.fetch ?? globalThis.fetch + this.#timeoutMs = positiveTimeout(options.timeoutMs, 10_000) + } + + async publishAgentCard(card: A2aAgentCard): Promise { + const canonical = A2aAgentCardSchema.parse(card) + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), this.#timeoutMs) + try { + const response = await this.#fetch(new URL('/v1/a2a/register', `${this.#baseUrl}/`), { + method: 'POST', + headers: { + accept: 'application/json', + authorization: `Bearer ${this.#token}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ agent_card: canonical }), + signal: controller.signal, + }) + const payload = await readJson(response) + if (response.status === 409) { + return await this.#resolveExistingCard(canonical, controller.signal) + } + if (!response.ok) { + throw new Error(`Relaycast rejected Factory persona card (${response.status}): ${errorDetail(payload)}`) + } + const data = asRecord(asRecord(payload)?.data) + const address = readString(data, 'relay_name', 'relayName', 'address') + if (!address) { + throw new Error('Relaycast accepted the Factory persona card without returning its relay address') + } + return { + name: canonical.name, + address, + ...(readString(data, 'certification') ? { certification: readString(data, 'certification') } : {}), + } + } catch (error) { + if (controller.signal.aborted) { + throw new Error(`Timed out publishing Factory persona card after ${this.#timeoutMs}ms`, { cause: error }) + } + throw error + } finally { + clearTimeout(timeout) + } + } + + async #resolveExistingCard( + canonical: A2aAgentCard, + signal: AbortSignal, + ): Promise { + const response = await this.#fetch(new URL('/v1/a2a/agents', `${this.#baseUrl}/`), { + method: 'GET', + headers: { + accept: 'application/json', + authorization: `Bearer ${this.#token}`, + }, + signal, + }) + const payload = await readJson(response) + if (!response.ok) { + throw new Error( + `Relaycast reported an existing Factory persona card but its record could not be verified ` + + `(${response.status}): ${errorDetail(payload)}`, + ) + } + + const records = directoryRows(payload) + let sameName = false + for (const value of records) { + const record = asRecord(value) + const parsed = A2aAgentCardSchema.safeParse(record?.agent_card ?? record?.agentCard) + if (!parsed.success || parsed.data.name !== canonical.name) continue + sameName = true + if (!isDeepStrictEqual(parsed.data, canonical)) continue + const address = readString(record, 'relay_name', 'relayName', 'address') + if (!address) { + throw new Error('Relaycast returned the existing Factory persona card without its relay address') + } + return { name: canonical.name, address, alreadyPublished: true } + } + + throw new Error(sameName + ? `Relaycast already has a different card for Factory persona "${canonical.name}"` + : `Relaycast reported a conflict but no card for Factory persona "${canonical.name}" was found`) + } +} + +function parsePersona(value: unknown): PersonaSpec { + const record = asRecord(value) + const intent = readString(record, 'intent') + if (!intent || !personaKitSpec.isIntent(intent)) { + throw new Error('Factory-hosted persona must declare a valid persona-kit intent') + } + return personaKitSpec.parsePersonaSpec(value, intent as PersonaIntent) +} + +function deriveAgentCardCompatibility(persona: PersonaSpec, options: DeriveAgentCardOptions): A2aAgentCard { + const integrationTags = Object.keys(persona.integrations ?? {}) + const skills = persona.skills.map((skill) => ({ + id: skill.id, + name: humanize(skill.id), + description: skill.description, + tags: unique([skill.source, ...integrationTags]), + })) + for (const [name, value] of Object.entries(persona.capabilities ?? {})) { + if (!capabilityEnabled(value)) continue + if (A2A_TRANSPORT_CAPABILITIES.has(name)) continue + const canonicalName = name === 'pullRequest' ? 'review' : name + const existing = skills.find((skill) => skill.id === canonicalName) + if (existing) { + existing.tags = unique([...(existing.tags ?? []), ...integrationTags]) + continue + } + skills.push({ + id: canonicalName, + name: humanize(canonicalName), + description: `Persona capability: ${humanize(canonicalName)}`, + tags: unique(integrationTags), + }) + } + + if (skills.length === 0) { + skills.push({ + id: persona.intent, + name: humanize(persona.intent), + description: persona.description, + tags: unique(integrationTags), + }) + } + + const relayName = typeof persona.relay === 'object' && persona.relay !== null + ? persona.relay.agentName + : undefined + return { + name: relayName ?? persona.id, + description: persona.description, + url: options.baseUrl, + version: options.version, + skills, + capabilities: { + streaming: capabilityEnabled(persona.capabilities?.streaming), + pushNotifications: capabilityEnabled(persona.capabilities?.pushNotifications), + }, + default_input_modes: options.inputModes ?? ['text/plain', 'application/json'], + default_output_modes: options.outputModes ?? ['text/plain', 'application/json'], + provider: { + organization: 'AgentWorkforce', + persona_id: persona.id, + intent: persona.intent, + tags: [...(persona.tags ?? [])], + }, + ...(options.documentationUrl ? { documentation_url: options.documentationUrl } : {}), + } +} + +function capabilityEnabled(value: unknown): boolean { + if (value === true) return true + if (value === false || value === undefined || value === null) return false + return typeof value === 'object' && !Array.isArray(value) && (value as { enabled?: unknown }).enabled !== false +} + +function humanize(value: string): string { + return value + .replace(/([a-z0-9])([A-Z])/gu, '$1 $2') + .replace(/[-_:]+/gu, ' ') + .replace(/\b\w/gu, (letter) => letter.toUpperCase()) +} + +function unique(values: string[]): string[] { + return [...new Set(values.map((value) => value.trim()).filter(Boolean))] +} + +async function readJson(response: Response): Promise { + const text = await response.text() + if (!text) return undefined + try { + return JSON.parse(text) as unknown + } catch (error) { + throw new Error('Relaycast card publication returned invalid JSON', { cause: error }) + } +} + +function errorDetail(payload: unknown): string { + const record = asRecord(payload) + const error = asRecord(record?.error) + return readString(error, 'message') ?? readString(record, 'message') ?? 'request failed' +} + +function directoryRows(payload: unknown): unknown[] { + if (Array.isArray(payload)) return payload + const record = asRecord(payload) + return Array.isArray(record?.data) ? record.data : [] +} + +function positiveTimeout(value: number | undefined, fallback: number): number { + return Number.isFinite(value) && (value ?? 0) > 0 ? Math.floor(value as number) : fallback +} + +function requiredText(value: string, label: string): string { + const normalized = value.trim() + if (!normalized) throw new Error(`${label} must be a non-empty string`) + return normalized +} + +function asRecord(value: unknown): Record | undefined { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined +} + +function readString(record: Record | undefined, ...keys: string[]): string | undefined { + if (!record) return undefined + for (const key of keys) { + const value = record[key] + if (typeof value === 'string' && value.trim()) return value.trim() + } + return undefined +} diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index f795e72..19bedf6 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -5,6 +5,7 @@ export type Capability = 'spawn:codex' | 'spawn:claude' | 'workflow:run' export type PreviewCapability = 'preview:tailscale-serve' export type NodeCapability = Capability | PreviewCapability export type RestartPolicy = import('@agent-relay/harness-driver').SpawnPtyInput['restartPolicy'] +export type A2aSkill = import('@relaycast/a2a').A2aSkill export type PreviewReference = { id: string @@ -97,6 +98,32 @@ export interface RosterEntry { nodes: Array<{ name: string; capabilities: NodeCapability[]; live: boolean }> } +export interface TeammateQuery { + /** Exact A2A skill id/name to match. */ + skill?: string + /** Exact agent- or skill-level tag to match. */ + tag?: string + /** Free-text directory query. */ + q?: string +} + +export interface TeammateAgent { + /** Directory identity shown to workers. */ + name: string + /** Human-readable card description, when the directory provides one. */ + description?: string + /** Canonical A2A skill records from the agent card. */ + skills: A2aSkill[] + /** A2A endpoint advertised by the directory. */ + url: string + kind: 'native' | 'a2a' + /** Relay/A2A target used for the discover -> engage hop. */ + address: string + tags: string[] + status?: string + certification?: string +} + export type AgentPidResolution = | { status: 'found'; pid: number } | { status: 'missing' } @@ -143,6 +170,8 @@ export interface FleetClient { }): Promise release(name: string, reason?: string): Promise roster(): Promise + /** Find addressable teammate agents by their published A2A cards. */ + discoverTeammates(query: TeammateQuery): Promise resolveAgentPid?(name: string): Promise protectedPids?(): Promise sendMessage(input: SendInput): Promise diff --git a/src/ports/index.ts b/src/ports/index.ts index e2db6c3..92b22c6 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -15,6 +15,7 @@ export type { Subscription, } from './mount' export type { + A2aSkill, AgentLifecycleSignal, AgentMessage, AgentPidResolution, @@ -32,6 +33,8 @@ export type { SendInput, SpawnInput, SpawnResult, + TeammateAgent, + TeammateQuery, } from './fleet' export type { GithubRead, diff --git a/src/testing/fakes.ts b/src/testing/fakes.ts index 64eaa06..b4109cd 100644 --- a/src/testing/fakes.ts +++ b/src/testing/fakes.ts @@ -10,6 +10,8 @@ import type { SendInput, SpawnInput, SpawnResult, + TeammateAgent, + TeammateQuery, SubscribeOptions, Subscription, Capability, @@ -204,6 +206,7 @@ export class FakeFleetClient implements FleetClient { readonly previewStarts: PreviewStartInput[] = [] readonly previewRemovals: PreviewReference[] = [] readonly previewSweeps: PreviewSweepInput[] = [] + readonly teammates: TeammateAgent[] = [] #agents = new Set() #tracked = new Map() @@ -244,6 +247,20 @@ export class FakeFleetClient implements FleetClient { this.#tracked.delete(name) } + async discoverTeammates(query: TeammateQuery): Promise { + const skill = query.skill?.trim().toLowerCase() + const tag = query.tag?.trim().toLowerCase() + const q = query.q?.trim().toLowerCase() + return this.teammates.filter((teammate) => { + if (skill && !teammate.skills.some((candidate) => + candidate.id?.toLowerCase() === skill || candidate.name.toLowerCase() === skill)) return false + if (tag && ![...teammate.tags, ...teammate.skills.flatMap((candidate) => candidate.tags ?? [])] + .some((candidate) => candidate.toLowerCase() === tag)) return false + if (q && !JSON.stringify(teammate).toLowerCase().includes(q)) return false + return true + }) + } + async createPreview(input: PreviewStartInput): Promise { this.previewStarts.push(structuredClone(input)) const httpsPort = input.preferredHttpsPort ?? 10_000 + this.previewStarts.length - 1 diff --git a/test/e2e/ask-a-teammate.test.ts b/test/e2e/ask-a-teammate.test.ts new file mode 100644 index 0000000..495d329 --- /dev/null +++ b/test/e2e/ask-a-teammate.test.ts @@ -0,0 +1,306 @@ +import { readFileSync } from 'node:fs' +import { createServer, type Server } from 'node:http' + +import { A2aAgentCardSchema } from '@relaycast/a2a' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import type { BrokerEvent, SendMessageInput, SpawnPtyInput } from '@agent-relay/harness-driver' + +import { ensureRelayBroker } from '../../src/fleet/ensure-relay-broker' +import { InternalFleetClient, type HarnessDriverClientLike } from '../../src/fleet/internal-fleet-client' +import { RelayFleetClient } from '../../src/fleet/relay-fleet-client' +import { askTeammate, RelaycastTeammateDirectory } from '../../src/fleet/teammates' +import { createFactoryNodeDefinition, parseFactoryNodeConfig } from '../../src/node/factory-node' +import { RelaycastAgentCardPublisher } from '../../src/node/factory-persona-card' +import { startFactoryNode } from '../../src/node/factory-node-runtime' + +type DirectoryRow = { + name: string + description?: string + skills: Array<{ id?: string; name: string; description?: string; tags?: string[] }> + tags: string[] + url: string + kind: 'native' | 'a2a' + status: string +} + +const openServers: Server[] = [] +afterEach(async () => { + await Promise.all(openServers.splice(0).map((server) => new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()) + }))) +}) + +describe('discover -> ask -> reply', () => { + it('discovers one skilled teammate, completes a relay round trip, and publishes a hosted persona card', async () => { + const directoryRows: DirectoryRow[] = [{ + name: 'infra-agent', + description: 'Watches production infrastructure.', + skills: [{ id: 'infra-watch', name: 'Infra Watch' }], + tags: ['operations'], + url: 'http://relay.local/v1/dm', + kind: 'native', + status: 'online', + }, { + name: 'review-agent', + description: 'Reviews code changes.', + skills: [{ id: 'code-review', name: 'Code Review' }], + tags: ['quality'], + url: 'http://relay.local/v1/dm', + kind: 'native', + status: 'online', + }] + const directoryRequests: URL[] = [] + const server = createServer(async (request, response) => { + const origin = `http://${request.headers.host}` + const url = new URL(request.url ?? '/', origin) + if (request.method === 'GET' && url.pathname === '/v1/a2a/directory') { + directoryRequests.push(url) + // Deliberately return every row. Factory must enforce exact skill/tag + // filtering too, otherwise this E2E returns both cards and fails. + json(response, 200, { ok: true, data: directoryRows }) + return + } + if (request.method === 'POST' && url.pathname === '/v1/a2a/register') { + const body = JSON.parse(await requestBody(request)) as { agent_card?: unknown } + const card = A2aAgentCardSchema.parse(body.agent_card) + const relayName = `ext-${card.name}-a1b2c3d4` + directoryRows.push({ + // Relaycast exposes the derived relay proxy identity as `name`; it + // does not add a separate address field to directory rows. + name: relayName, + skills: card.skills, + tags: Array.isArray(card.provider?.tags) + ? card.provider.tags.filter((value): value is string => typeof value === 'string') + : [], + url: `${origin}/a2a/rpc`, + kind: 'a2a', + status: 'active', + }) + json(response, 201, { + ok: true, + data: { relay_name: relayName, certification: 'level_1' }, + }) + return + } + json(response, 404, { ok: false }) + }) + openServers.push(server) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + if (!address || typeof address === 'string') throw new Error('mock directory did not bind') + const baseUrl = `http://127.0.0.1:${address.port}` + + const discoveryFleet = new RelayFleetClient({ + baseUrl, + workspaceKey: 'rk_live_test', + directoryFetch: fetch, + }) + const infra = await discoveryFleet.discoverTeammates({ skill: 'infra-watch' }) + expect(infra).toEqual([expect.objectContaining({ + name: 'infra-agent', + description: 'Watches production infrastructure.', + address: 'infra-agent', + url: 'http://relay.local/v1/dm', + kind: 'native', + skills: [expect.objectContaining({ id: 'infra-watch' })], + })]) + await expect(discoveryFleet.discoverTeammates({ skill: 'unknown-skill' })).resolves.toEqual([]) + expect(directoryRequests[0]?.searchParams.get('skill')).toBe('infra-watch') + + const broker = new InProcessRelayBroker(['factory-worker', 'infra-agent']) + const brokerHandle = await ensureRelayBroker({ + connect: () => { throw new Error('no existing broker') }, + spawn: async () => broker, + env: {}, + resolveWorkspaceKey: () => undefined, + }) + expect(brokerHandle.started).toBe(true) + const workerFleet = new InternalFleetClient({ + client: brokerHandle.client, + ownsBroker: brokerHandle.started, + teammateDirectory: new RelaycastTeammateDirectory({ + baseUrl, + token: 'rk_live_test', + }), + }) + const teammateFleet = new InternalFleetClient({ + client: brokerHandle.client, + }) + const cannedReply = 'The deploy is healthy; no intervention is needed.' + const questions: string[] = [] + const stopStub = teammateFleet.onAgentMessage((message) => { + if (message.from !== 'factory-worker' || message.target !== 'infra-agent') return + questions.push(message.body) + void teammateFleet.sendMessage({ + from: 'infra-agent', + to: 'factory-worker', + text: cannedReply, + }) + }) + + const roundTrip = await askTeammate(workerFleet, { + from: 'factory-worker', + question: 'Is the deploy healthy?', + skill: 'infra-watch', + timeoutMs: 1_000, + }) + expect(questions).toEqual(['Is the deploy healthy?']) + expect(roundTrip.teammate.name).toBe('infra-agent') + expect(roundTrip.reply).toMatchObject({ + from: 'infra-agent', + target: 'factory-worker', + body: cannedReply, + }) + + const persona = JSON.parse(readFileSync( + new URL('../../.agentworkforce/agents/factory-feature-guardian/persona.json', import.meta.url), + 'utf8', + )) as unknown + const definition = createFactoryNodeDefinition({ + config: parseFactoryNodeConfig({ + workspaceId: 'workspace-test', + capabilities: ['spawn:codex'], + clonePaths: { 'AgentWorkforce/factory': '/work/factory' }, + dryRun: false, + }), + name: 'factory-persona-node', + persona: { + persona, + baseUrl, + version: 'e2e', + }, + }) + expect(() => A2aAgentCardSchema.parse(definition.agentCard)).not.toThrow() + const running = startFactoryNode({ + definition, + connection: { nodeId: 'node-test', nodeToken: 'nt_live_test' }, + cardPublisher: new RelaycastAgentCardPublisher({ + baseUrl, + token: 'rk_live_test', + }), + serve(options) { + queueMicrotask(() => options.onRegistered?.({ + name: definition.name, + capabilities: Object.keys(definition.capabilities), + })) + return { stop: async () => {}, done: Promise.resolve() } + }, + }) + await expect(running.cardPublished).resolves.toMatchObject({ + name: 'factory-feature-guardian', + }) + await expect(discoveryFleet.discoverTeammates({ + skill: 'factory-feature-verification', + })).resolves.toEqual([expect.objectContaining({ + name: 'ext-factory-feature-guardian-a1b2c3d4', + address: 'ext-factory-feature-guardian-a1b2c3d4', + kind: 'a2a', + url: `${baseUrl}/a2a/rpc`, + })]) + + stopStub() + await teammateFleet.dispose() + await workerFleet.dispose() + expect(broker.shutdownCalls).toBe(1) + await discoveryFleet.dispose() + }) +}) + +class InProcessRelayBroker implements HarnessDriverClientLike { + readonly brokerPid = process.pid + readonly #agents: Array<{ name: string; cli?: string; pid?: number }> + readonly #events = new Set<(event: BrokerEvent) => void>() + readonly #deliveryEvents = new Set<(event: BrokerEvent) => void>() + readonly #exitEvents = new Set<(agent: { name: string; sessionId?: string }) => void>() + #sequence = 0 + shutdownCalls = 0 + + constructor(agentNames: string[]) { + this.#agents = agentNames.map((name) => ({ name, cli: 'stub', pid: process.pid })) + } + + async spawnPty(input: SpawnPtyInput) { + this.#agents.push({ name: input.name, cli: input.cli, pid: process.pid }) + return { name: input.name, session_ref: `session-${input.name}`, pid: process.pid } + } + + async release(name: string): Promise<{ name: string }> { + const index = this.#agents.findIndex((agent) => agent.name === name) + if (index >= 0) this.#agents.splice(index, 1) + return { name } + } + + async listAgents() { + return this.#agents.map((agent) => ({ ...agent })) + } + + async sendMessage(input: SendMessageInput): Promise<{ event_id: string; targets: string[] }> { + if (!this.#agents.some((agent) => agent.name === input.to)) { + throw new Error(`unknown recipient ${input.to}`) + } + const eventId = `event-${++this.#sequence}` + queueMicrotask(() => { + this.#emit({ + kind: 'delivery_injected', + name: input.to, + delivery_id: `delivery-${this.#sequence}`, + event_id: eventId, + }) + this.#emit({ + kind: 'relay_inbound', + event_id: `inbound-${this.#sequence}`, + from: input.from ?? 'factory', + target: input.to, + body: input.text, + }) + }) + return { event_id: eventId, targets: [input.to] } + } + + async sendInput(): Promise {} + connectEvents(): void {} + disconnect(): void {} + async shutdown(): Promise { this.shutdownCalls += 1 } + + onEvent(listener: (event: BrokerEvent) => void): () => void { + this.#events.add(listener) + return () => this.#events.delete(listener) + } + + addListener(event: 'agentExited', listener: (agent: { name: string; sessionId?: string }) => void): () => void + addListener(event: 'deliveryUpdate', listener: (event: BrokerEvent) => void): () => void + addListener( + event: 'agentExited' | 'deliveryUpdate', + listener: ((agent: { name: string; sessionId?: string }) => void) | ((event: BrokerEvent) => void), + ): () => void { + if (event === 'agentExited') { + const exitListener = listener as (agent: { name: string; sessionId?: string }) => void + this.#exitEvents.add(exitListener) + return () => this.#exitEvents.delete(exitListener) + } + const deliveryListener = listener as (event: BrokerEvent) => void + this.#deliveryEvents.add(deliveryListener) + return () => this.#deliveryEvents.delete(deliveryListener) + } + + #emit(event: BrokerEvent): void { + for (const listener of this.#events) listener(event) + for (const listener of this.#deliveryEvents) listener(event) + } +} + +function requestBody(request: import('node:http').IncomingMessage): Promise { + return new Promise((resolve, reject) => { + let body = '' + request.setEncoding('utf8') + request.on('data', (chunk) => { body += chunk }) + request.on('end', () => resolve(body)) + request.on('error', reject) + }) +} + +function json(response: import('node:http').ServerResponse, status: number, body: unknown): void { + response.writeHead(status, { 'content-type': 'application/json' }) + response.end(JSON.stringify(body)) +} diff --git a/vitest.config.ts b/vitest.config.ts index 18c5c82..20104d9 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -2,7 +2,7 @@ import { defineConfig } from 'vitest/config' export default defineConfig({ test: { - include: ['src/**/*.test.ts', '.agentworkforce/agents/**/*.test.ts'], + include: ['src/**/*.test.ts', 'test/e2e/**/*.test.ts', '.agentworkforce/agents/**/*.test.ts'], exclude: ['node_modules/**', 'dist/**', 'out/**'], }, })