diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index 1f075d7a..baced2e1 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -203,7 +203,9 @@ vi.mock('@agent-relay/harness-driver', () => ({ })) vi.mock('./pear-fleet-node', () => ({ - startPearFleetSidecar: fleetNodeMock.startPearFleetSidecar + startPearFleetSidecar: fleetNodeMock.startPearFleetSidecar, + pearFleetProviderName: (options: { brokerName?: string; projectId?: string }) => + `${options.brokerName || options.projectId || 'pear'}-local-fleet` })) vi.mock('./auth', () => ({ diff --git a/src/main/broker.ts b/src/main/broker.ts index dd30f865..025d4fab 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -16,7 +16,7 @@ import { type InboundDeliveryMode, type PendingRelayMessage } from '@agent-relay/harness-driver' -import { AgentRelay, type RelayMessage } from '@agent-relay/sdk' +import { AgentRelay, RelayPlacementError, type RelayMessage } from '@agent-relay/sdk' import { getAccessToken, getApiUrl } from './auth' import { assertDirectory } from './path-utils' import { toErrorMessage } from './errors' @@ -71,13 +71,26 @@ import { resolveCommandOnPath, resolvePackageBin } from './mcp-command' -import { startPearFleetSidecar, type RunningPearFleetSidecar } from './pear-fleet-node' +import { startPearFleetSidecar, pearFleetProviderName, type RunningPearFleetSidecar } from './pear-fleet-node' import { isObserverStreamEnabled, ObserverStreamManager, ObserverStreamUnsupportedError } from './observer-stream' import { getObserverStreamCursor, setObserverStreamCursor } from './store' +import { + BrokerPlacementError, + buildPlacementMessage, + placementRequesterName, + toBrokerNodeSummary +} from './placement' +import type { + BrokerPlaceAgentInput, + BrokerPlaceAgentResult, + BrokerNodeSummary +} from '../shared/types/ipc' + +export { BrokerPlacementError } from './placement' function isShellLikeCommand(cli: string): boolean { const normalized = basename(cli).toLowerCase() @@ -1089,6 +1102,11 @@ interface BrokerSession { leaseTimer?: ReturnType fleetSidecar?: RunningPearFleetSidecar fleetSidecarCwd?: string + // Agent-scoped relay client used to invoke placement (#411). Lazily created on + // first placeAgent/listNodes: workspace key from the broker session + a + // dedicated `pear-requester-` agent identity (placement.spawn -> + // commands.invoke requires an agent-scoped connection). Cleared in dropSession. + placementRelay?: AgentRelay operationQueue: BrokerOperationQueue } @@ -2544,6 +2562,131 @@ export class BrokerManager { return normalized } + // Lazily build (and cache on the session) an agent-scoped relay client for + // placement. placement.spawn -> commands.invoke requires an agent-scoped + // connection, so a workspace-key-only client (as reconcileMessages uses) is + // insufficient: register/rotate a dedicated `pear-requester-` + // identity and construct the client with its agent token. + private async getPlacementRelay(session: BrokerSession): Promise { + if (session.placementRelay) return session.placementRelay + + const meta = await session.client.getSession() + const workspaceKey = meta.workspace_key + if (!workspaceKey) { + throw new Error('Broker session does not expose a Relay workspace key') + } + const baseUrl = normalizeRelaycastBaseUrl(process.env.RELAYCAST_BASE_URL || process.env.RELAY_BASE_URL) + const workspaceRelay = new AgentRelay({ workspaceKey, ...(baseUrl ? { baseUrl } : {}) }) + const requesterName = placementRequesterName(session.projectId) + // registerOrRotate adopts an existing `pear-requester-` identity + // (rotating its token) so restarts don't strand orphan agents; fall back to + // register on backends without rotation support. + const register = workspaceRelay.agents.registerOrRotate ?? workspaceRelay.agents.register + const registration = await register.call(workspaceRelay.agents, { + name: requesterName, + type: 'agent', + metadata: { pearRequester: true, projectId: session.projectId } + }) + const relay = new AgentRelay({ + workspaceKey, + agentToken: registration.token, + ...(baseUrl ? { baseUrl } : {}) + }) + session.placementRelay = relay + return relay + } + + private localFleetNodeName(session: BrokerSession): string { + return pearFleetProviderName({ + projectId: session.projectId, + cwd: session.cwd, + brokerName: session.name + }) + } + + /** + * Placement requester (#411). Dispatch a spawn onto an eligible fleet node via + * the relay placement engine. `input.node` omitted → any eligible least-loaded + * node; set → that exact node ('self' = this machine). Returns the node the + * agent landed on and whether this machine owns its PTY (`local`). A remote + * placement is reachable over relay chat only — its raw terminal has no relay + * transport yet (acceptance #2b, upstream-blocked). + */ + async placeAgent(projectId: string, input: BrokerPlaceAgentInput): Promise { + const normalizedProjectId = projectId.trim() + if (!normalizedProjectId) throw new Error('Project id is required') + const cli = spawnCliLabel(input.cli) + if (!cli) throw new Error('cli is required for placement') + + const session = this.getSessionForProject(normalizedProjectId) + const relay = await this.getPlacementRelay(session) + const capability = `spawn:${cli}` + const selfNodeName = this.localFleetNodeName(session) + const requestedNode = input.node?.trim() || undefined + + // Dedupe the placed name against the shared workspace roster — a remote + // landing shares the workspace, so a name collision there is workspace-wide + // even though a single broker would accept it locally. + const existingNames = new Set( + (await relay.agents.list().catch(() => [])).map((agent) => agent.name) + ) + const name = getAvailableAgentName(input.name?.trim() || `${cli}-1`, existingNames) + // Any-node placement fails fast (a clear message, never a silent hang); + // an explicit target queues up to the TTL so a briefly-offline node recovers. + const failFast = input.failFast ?? !requestedNode + + try { + const ack = await relay.messaging.placement.spawn({ + capability, + ...(requestedNode ? { node: requestedNode } : {}), + ...(requestedNode === 'self' ? { selfNodeName } : {}), + ...(input.repo?.trim() ? { repo: input.repo.trim() } : {}), + input: { + name, + ...(input.task?.trim() ? { task: input.task.trim() } : {}), + ...(input.model?.trim() ? { model: input.model.trim() } : {}) + }, + failFast + }) + const landedNode = ack.node.name + const nodeId = ack.node.nodeId ?? ack.node.id + return { + name, + node: landedNode, + ...(nodeId ? { nodeId } : {}), + invocationId: ack.invocationId, + queued: ack.placement.queued, + local: landedNode === selfNodeName + } + } catch (err) { + if (err instanceof RelayPlacementError) { + throw new BrokerPlacementError(err.code, buildPlacementMessage(err), { + capability: err.capability, + node: err.node, + repo: err.repo + }) + } + throw err + } + } + + /** + * Fleet node roster for the spawn/node-picker UI (#411). Lists nodes visible + * in the project's relay workspace, optionally filtered to those advertising a + * capability (e.g. `spawn:claude`), flagging this machine's own node. + */ + async listNodes(projectId: string, capability?: string): Promise { + const normalizedProjectId = projectId.trim() + if (!normalizedProjectId) throw new Error('Project id is required') + const session = this.getSessionForProject(normalizedProjectId) + const relay = await this.getPlacementRelay(session) + const selfNodeName = this.localFleetNodeName(session) + const nodes = await relay.nodes.list( + capability?.trim() ? { capability: capability.trim() } : undefined + ) + return nodes.map((node) => toBrokerNodeSummary(node, selfNodeName)) + } + private attachClient( sessionKey: string, client: AgentRelayClient, @@ -4342,6 +4485,10 @@ export class BrokerManager { session.unsubEvent() if (session.leaseTimer) clearInterval(session.leaseTimer) + // Drop the cached placement requester client; a session reconnecting to a + // different workspace would otherwise reuse an agent token scoped to the old + // workspace (mirrors the observer-token invalidation above). + session.placementRelay = undefined void this.stopSessionFleetSidecar(session) if (options.disconnectOnly) { const disconnect = (session.client as { disconnect?: () => void }).disconnect diff --git a/src/main/ipc-handlers.ts b/src/main/ipc-handlers.ts index c34f61c3..ef749b74 100644 --- a/src/main/ipc-handlers.ts +++ b/src/main/ipc-handlers.ts @@ -21,7 +21,7 @@ import { addProjectIntegration, removeProjectIntegration } from './store' -import { brokerManager, isCommandAvailableWithAugmentedPath } from './broker' +import { brokerManager, isCommandAvailableWithAugmentedPath, BrokerPlacementError } from './broker' import * as git from './git' import * as filesystem from './filesystem' import * as auth from './auth' @@ -37,6 +37,9 @@ import { findProjectForPath, projectContainsPath } from './cli' import type { BrokerReconcileMessagesInput, BrokerSpawnAgentResult, + BrokerPlaceAgentInput, + BrokerPlaceAgentOutcome, + BrokerNodeSummary, FactoryAgentStatus, FactoryConfigReadResult, FactoryIssueStatus, @@ -762,6 +765,33 @@ export function registerIpcHandlers(): void { return toBrokerSpawnAgentResult(result) }) + // Placement requester (#411): dispatch a spawn onto an eligible fleet node. + // Placement errors (no eligible node, capability mismatch, queue full, unmapped + // repo) are returned as structured outcomes so the UI can show a clear message + // and never hang; unexpected failures still reject. + ipcMain.handle('broker:place-agent', async (_, projectId: string, input: BrokerPlaceAgentInput): Promise => { + try { + const result = await brokerManager.placeAgent(projectId, input) + integrationEventBridge.invalidateProjectAgentCache(projectId) + return { status: 'placed', result } + } catch (err) { + if (err instanceof BrokerPlacementError) { + return { + status: 'error', + code: err.code, + message: err.message, + ...(err.node ? { node: err.node } : {}), + ...(err.repo ? { repo: err.repo } : {}) + } + } + throw err + } + }) + + ipcMain.handle('broker:list-nodes', async (_, projectId: string, capability?: string): Promise => { + return brokerManager.listNodes(projectId, capability) + }) + ipcMain.handle('broker:list-personas', async (_, projectId: string, cwd?: string) => { const normalizedProjectId = projectId.trim() const personaCwd = cwd?.trim() diff --git a/src/main/pear-fleet-node.ts b/src/main/pear-fleet-node.ts index 0f343b6e..036319fc 100644 --- a/src/main/pear-fleet-node.ts +++ b/src/main/pear-fleet-node.ts @@ -377,7 +377,7 @@ export function startPearFleetSidecar(options: PearFleetSidecarOptions): Running } } -function pearFleetProviderName(options: PearFleetNodeOptions): string { +export function pearFleetProviderName(options: PearFleetNodeOptions): string { const rawName = `${options.brokerName || options.projectId || 'pear'}-local-fleet` return rawName.replace(/[^\w.-]+/gu, '-').replace(/^-+|-+$/gu, '') || 'pear-local-fleet' } diff --git a/src/main/placement-helpers.test.ts b/src/main/placement-helpers.test.ts new file mode 100644 index 00000000..d6aef6f5 --- /dev/null +++ b/src/main/placement-helpers.test.ts @@ -0,0 +1,97 @@ +import { describe, it, expect } from 'vitest' +import { RelayPlacementError, type RelayNode } from '@agent-relay/sdk' +import { + buildPlacementMessage, + placementRequesterName, + toBrokerNodeSummary +} from './placement' + +function placementError(code: RelayPlacementError['code'], ctx: { capability?: string; node?: string; repo?: string } = {}): RelayPlacementError { + return new RelayPlacementError(code, `raw ${code}`, { + capability: ctx.capability ?? 'spawn:claude', + node: ctx.node, + repo: ctx.repo, + attempts: 1 + }) +} + +describe('buildPlacementMessage', () => { + it('names the offending node and cli for capability_mismatch', () => { + const message = buildPlacementMessage(placementError('capability_mismatch', { node: 'gpu-box-1' })) + expect(message).toBe('Node "gpu-box-1" can\'t run claude.') + }) + + it('falls back to a generic capability message when no node is named', () => { + const message = buildPlacementMessage(placementError('capability_mismatch')) + expect(message).toBe('No node can run claude.') + }) + + it('reports queue saturation for placement_queue_full', () => { + expect(buildPlacementMessage(placementError('placement_queue_full'))).toMatch(/try again/i) + }) + + it('reports no eligible node for placement_ttl_expired (never a hang)', () => { + expect(buildPlacementMessage(placementError('placement_ttl_expired'))).toBe( + 'No node advertises claude right now.' + ) + }) + + it('names the repo for unmapped_repo', () => { + expect(buildPlacementMessage(placementError('unmapped_repo', { repo: 'pear' }))).toBe( + 'No node has repo "pear" checked out.' + ) + }) + + it('strips the spawn: prefix from the capability in messages', () => { + const message = buildPlacementMessage(placementError('placement_ttl_expired', { capability: 'spawn:codex' })) + expect(message).toContain('codex') + expect(message).not.toContain('spawn:') + }) +}) + +describe('placementRequesterName', () => { + it('derives a sanitized, workspace-safe requester identity per project', () => { + expect(placementRequesterName('project-1')).toBe('pear-requester-project-1') + }) + + it('replaces path/space characters that a workspace name cannot carry', () => { + expect(placementRequesterName('a/b c:d')).toBe('pear-requester-a-b-c-d') + }) + + it('never returns an empty name', () => { + expect(placementRequesterName('')).toBe('pear-requester') + }) +}) + +describe('toBrokerNodeSummary', () => { + const baseNode: RelayNode = { + name: 'other-node', + status: 'online', + live: true, + load: 2, + activeAgents: 1, + maxAgents: 4, + capabilities: [{ name: 'spawn:claude' }, { name: 'spawn:codex' }], + repoKeys: ['pear'], + tags: ['pear', 'local'] + } as RelayNode + + it('flattens capabilities and preserves liveness/load for the picker', () => { + const summary = toBrokerNodeSummary(baseNode, 'my-self-node') + expect(summary.capabilities).toEqual(['spawn:claude', 'spawn:codex']) + expect(summary.live).toBe(true) + expect(summary.load).toBe(2) + expect(summary.activeAgents).toBe(1) + expect(summary.isSelf).toBe(false) + }) + + it('flags this machine when the node name matches the local fleet node', () => { + const summary = toBrokerNodeSummary({ ...baseNode, name: 'my-self-node' }, 'my-self-node') + expect(summary.isSelf).toBe(true) + }) + + it('treats an absent live flag as offline', () => { + const summary = toBrokerNodeSummary({ ...baseNode, live: undefined }, 'my-self-node') + expect(summary.live).toBe(false) + }) +}) diff --git a/src/main/placement.ts b/src/main/placement.ts new file mode 100644 index 00000000..7bcedaea --- /dev/null +++ b/src/main/placement.ts @@ -0,0 +1,78 @@ +/** + * Placement (issue #411, requester side) — pure, electron-free helpers shared by + * the broker requester path and the isolated E2E harness. A placed agent is + * dispatched via the relay placement engine (`messaging.placement.spawn`) rather + * than the local broker's PTY driver; these helpers translate the SDK's + * RelayPlacementError into a user-facing message and shape the node roster for + * the picker UI. Kept out of broker.ts so a Node script (the Rung-1 gate) can + * import them without pulling in electron. + */ +import type { RelayPlacementError, RelayNode } from '@agent-relay/sdk' +import type { BrokerPlacementErrorCode, BrokerNodeSummary } from '../shared/types/ipc' + +export class BrokerPlacementError extends Error { + readonly code: BrokerPlacementErrorCode + readonly capability?: string + readonly node?: string + readonly repo?: string + + constructor( + code: BrokerPlacementErrorCode, + message: string, + ctx: { capability?: string; node?: string; repo?: string } = {} + ) { + super(message) + this.name = 'BrokerPlacementError' + this.code = code + this.capability = ctx.capability + this.node = ctx.node + this.repo = ctx.repo + } +} + +export function placementRequesterName(projectId: string): string { + const raw = `pear-requester-${projectId}` + return raw.replace(/[^\w.-]+/gu, '-').replace(/^-+|-+$/gu, '').slice(0, 64) || 'pear-requester' +} + +export function humanCapability(capability: string): string { + return capability.startsWith('spawn:') ? capability.slice('spawn:'.length) : capability +} + +// The four RelayPlacementError codes each map to a clear, actionable message so +// the spawn UI never shows a raw SDK string or — worse — hangs silently (#411 +// fallback requirement). +export function buildPlacementMessage(err: RelayPlacementError): string { + switch (err.code) { + case 'capability_mismatch': + return err.node + ? `Node "${err.node}" can't run ${humanCapability(err.capability)}.` + : `No node can run ${humanCapability(err.capability)}.` + case 'placement_queue_full': + return 'Too many pending placements right now — try again in a moment.' + case 'placement_ttl_expired': + return `No node advertises ${humanCapability(err.capability)} right now.` + case 'unmapped_repo': + return err.repo + ? `No node has repo "${err.repo}" checked out.` + : 'No node maps the requested repo.' + default: + return err.message + } +} + +export function toBrokerNodeSummary(node: RelayNode, selfNodeName: string): BrokerNodeSummary { + const nodeId = node.nodeId ?? node.id + return { + name: node.name, + ...(nodeId ? { nodeId } : {}), + live: Boolean(node.live), + ...(typeof node.load === 'number' ? { load: node.load } : {}), + ...(typeof node.activeAgents === 'number' ? { activeAgents: node.activeAgents } : {}), + ...(typeof node.maxAgents === 'number' ? { maxAgents: node.maxAgents } : {}), + capabilities: node.capabilities.map((cap) => cap.name), + ...(node.repoKeys ? { repoKeys: node.repoKeys } : {}), + ...(node.tags ? { tags: node.tags } : {}), + isSelf: node.name === selfNodeName + } +} diff --git a/src/preload/index.ts b/src/preload/index.ts index 407fb576..c8e22ce1 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -21,6 +21,9 @@ import type { BrokerSetTerminalModeResult, BrokerSpawnAgentInput, BrokerSpawnAgentResult, + BrokerPlaceAgentInput, + BrokerPlaceAgentOutcome, + BrokerNodeSummary, BrokerStatusEvent, BrokerTerminalSnapshot, BrokerTerminalSnapshotFormat, @@ -257,6 +260,10 @@ const api = { connectCloud: () => invoke('broker:connect-cloud'), spawnAgent: (projectId: string, input: BrokerSpawnAgentInput) => invoke('broker:spawn-agent', projectId, input), + placeAgent: (projectId: string, input: BrokerPlaceAgentInput) => + invoke('broker:place-agent', projectId, input), + listNodes: (projectId: string, capability?: string) => + invoke('broker:list-nodes', projectId, capability), listPersonas: (projectId: string, cwd?: string) => invoke('broker:list-personas', projectId, cwd), spawnPersona: (projectId: string, personaId: string) => diff --git a/src/renderer/src/components/sidebar/SpawnAgentDialog.tsx b/src/renderer/src/components/sidebar/SpawnAgentDialog.tsx index 7a4b0f7e..435f648a 100644 --- a/src/renderer/src/components/sidebar/SpawnAgentDialog.tsx +++ b/src/renderer/src/components/sidebar/SpawnAgentDialog.tsx @@ -2,11 +2,15 @@ import type React from 'react' import { useCallback, useEffect, useRef, useState } from 'react' import { Loader2, X } from 'lucide-react' import { ClaudeIcon, CodexIcon, GrokIcon, OpenCodeIcon } from '@/components/common/AgentIcons' -import { SPAWN_AGENT_CLI_INSTALL_COMMANDS, listProjectPersonas, spawnProjectAgent, spawnProjectPersona, type SpawnAgentCli } from '@/lib/spawn-agent' -import { pear, type WorkforcePersona } from '@/lib/ipc' +import { SPAWN_AGENT_CLI_INSTALL_COMMANDS, listProjectPersonas, spawnProjectAgent, spawnProjectPersona, placeProjectAgent, type SpawnAgentCli } from '@/lib/spawn-agent' +import { pear, type WorkforcePersona, type BrokerNodeSummary } from '@/lib/ipc' import { useProjectStore, type ProjectRoot } from '@/stores/project-store' import { useUIStore } from '@/stores/ui-store' +// Sentinel for the "any eligible node" placement option (distinct from '' = +// this Mac, and from a concrete node name). +const ANY_NODE = '__any__' + const AGENT_OPTIONS: Array<{ cli: SpawnAgentCli; label: string; Icon: typeof ClaudeIcon }> = [ { cli: 'claude', label: 'Claude', Icon: ClaudeIcon }, { cli: 'codex', label: 'Codex', Icon: CodexIcon }, @@ -25,6 +29,12 @@ export function SpawnAgentDialog(): React.ReactNode { const [error, setError] = useState(null) const [cliAvailability, setCliAvailability] = useState>>({}) const [selectedRootId, setSelectedRootId] = useState(null) + // Placement (#411): '' = this Mac (direct local spawn), ANY_NODE = any eligible + // least-loaded node, or a specific remote node name. Only remote/any go through + // the placement engine; the local default stays on the proven direct path. + const [nodes, setNodes] = useState([]) + const [selectedNode, setSelectedNode] = useState('') + const [remoteNotice, setRemoteNotice] = useState(null) const project = useProjectStore((s) => s.getActiveProject()) const defaultRoot = useProjectStore((s) => s.getActiveRoot()) const selectedRoot = project?.roots.find((r) => r.id === selectedRootId) @@ -49,6 +59,18 @@ export function SpawnAgentDialog(): React.ReactNode { } }, [project, root?.id, selectedRootId]) + // Load the fleet node roster for the placement picker. Best-effort: a broker + // that isn't up yet just yields an empty list (local-only spawn still works). + useEffect(() => { + if (!project) return + let cancelled = false + void pear.broker.listNodes(project.id).then( + (roster) => { if (!cancelled) setNodes(roster) }, + () => { if (!cancelled) setNodes([]) } + ) + return () => { cancelled = true } + }, [project?.id]) + useEffect(() => { let cancelled = false const clis: SpawnAgentCli[] = ['claude', 'codex', 'opencode'] @@ -128,10 +150,27 @@ export function SpawnAgentDialog(): React.ReactNode { spawnRequestRef.current = true setError(null) + setRemoteNotice(null) setSpawningCli(cli) try { - await spawnProjectAgent(project, cli, customName, root, customModel) - closeDialog() + if (selectedNode === '') { + // This Mac — proven direct local spawn path (unchanged). + await spawnProjectAgent(project, cli, customName, root, customModel) + closeDialog() + } else { + // Placement engine: any eligible node, or a specific remote node. + const targetNode = selectedNode === ANY_NODE ? undefined : selectedNode + const placed = await placeProjectAgent(project, cli, targetNode, customName, customModel, root) + if (placed.local) { + closeDialog() + } else { + // Remote landing — no terminal view yet (upstream gap). Keep the dialog + // open with a clear chat-first message rather than opening a broken pane. + setRemoteNotice( + `Placed “${placed.name}” on ${placed.node}. It’s reachable via chat — a terminal view for remote nodes isn’t available yet (upstream gap).` + ) + } + } } catch (err) { setError(err instanceof Error ? err.message : String(err)) } finally { @@ -207,6 +246,31 @@ export function SpawnAgentDialog(): React.ReactNode { )}
{root?.path || project.rootPath}
+
+ + + {selectedNode !== '' && ( +

+ Remote agents run on another node and are reachable via chat; a terminal view for remote nodes isn’t available yet. +

+ )} +
diff --git a/src/renderer/src/lib/ipc-mock.ts b/src/renderer/src/lib/ipc-mock.ts index 128d7047..7d0bda6f 100644 --- a/src/renderer/src/lib/ipc-mock.ts +++ b/src/renderer/src/lib/ipc-mock.ts @@ -25,6 +25,9 @@ import type { BrokerSetTerminalModeResult, BrokerSpawnAgentInput, BrokerSpawnAgentResult, + BrokerPlaceAgentInput, + BrokerPlaceAgentOutcome, + BrokerNodeSummary, BrokerStatusEvent, BurnAgentBreakdown, BurnAgentInput, @@ -925,6 +928,36 @@ export const pearMock: PearAPI = { } satisfies AgentSpawnedEvent) return { name: agent.name, runtime: agent.runtime || 'mock', cli: agent.cli } }, + placeAgent: async (projectId: string, input: BrokerPlaceAgentInput): Promise => { + // Mock placement lands the agent locally so the spawn UI can be exercised + // without a real fleet; a real cross-node landing needs the isolated E2E. + const result = await pearMock.broker.spawnAgent(projectId, { + name: input.name || `${input.cli}-1`, + cli: input.cli, + ...(input.model ? { model: input.model } : {}), + ...(input.task ? { task: input.task } : {}) + }) + return { + status: 'placed', + result: { + name: result.name, + node: input.node && input.node !== 'self' ? input.node : 'mock-local-node', + invocationId: `mock-inv-${result.name}`, + queued: false, + local: !input.node || input.node === 'self' + } + } + }, + listNodes: async (): Promise => [ + { + name: 'mock-local-node', + live: true, + load: 0, + activeAgents: 0, + capabilities: ['spawn:claude', 'spawn:codex', 'spawn:opencode', 'spawn:grok'], + isSelf: true + } + ], listPersonas: async (): Promise => [], spawnPersona: async (projectId: string, personaId: string) => pearMock.broker.spawnAgent(projectId, { name: personaId, cli: 'codex' }), diff --git a/src/renderer/src/lib/spawn-agent.ts b/src/renderer/src/lib/spawn-agent.ts index 17c2c6a0..57a0400f 100644 --- a/src/renderer/src/lib/spawn-agent.ts +++ b/src/renderer/src/lib/spawn-agent.ts @@ -115,6 +115,63 @@ export async function spawnProjectAgent( return name } +export interface PlaceAgentUiResult { + name: string + node: string + /** True when the agent landed on this machine (has a local terminal). */ + local: boolean +} + +/** + * Place an agent on a fleet node via the relay placement engine (#411). `node` + * omitted/'' → any eligible least-loaded node (may resolve to this Mac); a node + * name targets that node; 'self' targets this machine. A local landing is + * integrated exactly like a direct spawn (terminal attached + tracked); a remote + * landing has no local PTY — its raw terminal has no relay transport yet + * (acceptance #2b), so it is surfaced chat-first and no terminal tab is opened. + * Placement failures throw with a clear, user-facing message (never a hang). + */ +export async function placeProjectAgent( + project: Project, + cli: SpawnAgentCli, + node: string | undefined, + customName?: string, + customModel?: string, + rootOverride?: ProjectRoot +): Promise { + await ensureLocalBroker(project, rootOverride) + const root = rootOverride ?? useProjectStore.getState().getActiveRoot() + if (!root?.pathExists) { + throw new Error(`Project root not found: ${root?.path || project.rootPath}`) + } + + const outcome = await pear.broker.placeAgent(project.id, { + cli, + ...(node && node.trim() ? { node: node.trim() } : {}), + ...(customName?.trim() ? { name: customName.trim() } : {}), + ...(customModel?.trim() ? { model: customModel.trim() } : {}) + }) + if (outcome.status === 'error') { + // A clear per-code message from the main process — surface it verbatim. + throw new Error(outcome.message) + } + + const { result } = outcome + const agentStore = useAgentStore.getState() + if (result.local) { + // Landed on this Mac — same integration as a direct local spawn. + await pear.broker.attachTerminal({ projectId: project.id, name: result.name, mode: 'passthrough' }) + agentStore.trackSpawnedAgent(result.name, project.id, root.id, cli, root.path) + agentStore.setActiveAgentKey(getAgentKey(project.id, result.name)) + useUIStore.getState().openTab({ kind: 'agents', projectId: project.id }) + } else { + // Remote landing — no local PTY. Reachable over relay chat; do NOT open a + // (would-be broken) terminal tab. The caller surfaces a chat-first message. + useUIStore.getState().openTab({ kind: 'agents', projectId: project.id }) + } + return { name: result.name, node: result.node, local: result.local } +} + export async function listProjectPersonas(project: Project, rootOverride?: ProjectRoot): Promise { if (rootOverride && !rootOverride.pathExists) { return [] diff --git a/src/shared/types/ipc.ts b/src/shared/types/ipc.ts index 2525d719..7fd3a2c0 100644 --- a/src/shared/types/ipc.ts +++ b/src/shared/types/ipc.ts @@ -412,6 +412,81 @@ export type TermFidelityCorpusResult = | { dumped: true; path: string } | { dumped: false; reason: 'rate-limited' | 'failed' } +/** + * Request to place an agent on a fleet node via the relay placement engine + * (issue #411 — the requester side). `node` omitted → any eligible, least-loaded + * node. `node` set → that exact node ('self' resolves to this machine's fleet + * node). A placed agent is reachable over relay messaging (chat); its raw + * terminal is owned by the target node's broker (see BrokerPlaceAgentResult). + */ +export interface BrokerPlaceAgentInput { + cli: string + name?: string + /** Exact target node name, or 'self' for this machine. Omit for any-node. */ + node?: string + /** Repo label that must be present in the selected node's repo map. */ + repo?: string + model?: string + task?: string + /** + * Fail immediately with a clear error instead of queueing when no eligible + * node exists. Defaults to true for any-node placement (no silent hang) and + * false for an explicit target (a briefly-offline node queues up to TTL). + */ + failFast?: boolean +} + +export type BrokerPlacementErrorCode = + | 'capability_mismatch' + | 'placement_queue_full' + | 'placement_ttl_expired' + | 'unmapped_repo' + +export interface BrokerPlaceAgentResult { + /** Agent name as placed (may be de-duplicated from the requested name). */ + name: string + /** Node the agent landed on. */ + node: string + nodeId?: string + /** Relay action invocation id for the spawn dispatch. */ + invocationId: string + /** True when the placement had to queue before an eligible node appeared. */ + queued: boolean + /** + * Whether this machine owns the placed agent's PTY. True only when the agent + * landed on the local ('self') node — remote placements have no raw-terminal + * transport over relay yet (acceptance #2b, upstream-blocked), so the UI must + * present a chat-first surface rather than a terminal pane. + */ + local: boolean +} + +/** + * Placement outcome surfaced to the renderer. A placement error is a normal, + * expected result (no eligible node, capability mismatch, etc.) — returned as + * structured data rather than a thrown IPC error so the spawn UI can render a + * clear per-code message and never hang. Unexpected failures still throw. + */ +export type BrokerPlaceAgentOutcome = + | { status: 'placed'; result: BrokerPlaceAgentResult } + | { status: 'error'; code: BrokerPlacementErrorCode; message: string; node?: string; repo?: string } + +/** A fleet node as surfaced to the spawn/node-picker UI. */ +export interface BrokerNodeSummary { + name: string + nodeId?: string + live: boolean + load?: number + activeAgents?: number + maxAgents?: number + /** Advertised capability names, e.g. 'spawn:claude'. */ + capabilities: string[] + repoKeys?: string[] + tags?: string[] + /** True when this node is this machine's own local fleet node. */ + isSelf?: boolean +} + export interface WorkforcePersona { id: string description?: string @@ -1000,6 +1075,8 @@ export interface PearAPI { ) => Promise<{ removed: string[] }> connectCloud: () => Promise spawnAgent: (projectId: string, input: BrokerSpawnAgentInput) => Promise + placeAgent: (projectId: string, input: BrokerPlaceAgentInput) => Promise + listNodes: (projectId: string, capability?: string) => Promise listPersonas: (projectId: string, cwd?: string) => Promise spawnPersona: (projectId: string, personaId: string) => Promise attachTerminal: (input: BrokerAttachTerminalInput) => Promise diff --git a/tests/placement/README.md b/tests/placement/README.md new file mode 100644 index 00000000..b5e27de3 --- /dev/null +++ b/tests/placement/README.md @@ -0,0 +1,56 @@ +# Placement requester E2E (issue #411) + +Rung-1 gate for the fleet placement **requester** side. Stands up two isolated +agent-relay brokers on this machine, enrolls broker **B** as a pear fleet node +(advertising `spawn:claude`), then — from a requester exactly as +`BrokerManager.placeAgent` does — places a **real** claude agent on B and proves +the mechanics acceptance #411 requires. + +## Run + +```sh +npx tsx tests/placement/rung1-cross-node.mts +# keep temp brokers/state for inspection: +PLACEMENT_E2E_KEEP=1 npx tsx tests/placement/rung1-cross-node.mts +# hard-abort (do not degrade) if the target workspace isn't hermetic: +PLACEMENT_E2E_REQUIRE_HERMETIC=1 npx tsx tests/placement/rung1-cross-node.mts +``` + +Requires: `claude` installed + authenticated; the agent-relay-broker binary +resolvable (auto-resolved, or `AGENT_RELAY_BIN`); relay/cloud auth available to +the spawned broker (same as the live app). + +## What it proves + +| Check | Acceptance | +|---|---| +| placement lands on remote node B (`ack.node==B`, invocation completed) | #1 cross-node | +| target node B exposes the placed agent PTY (HTTP 200) | transport | +| **requester broker A does NOT own the PTY (HTTP 404)** | **#2b — empirical proof the raw terminal has no relay transport (upstream gap)** | +| placed agent reachable over relay chat (round-trip) | #2a chat reachability | +| no-eligible → fails fast with a clear message | #4 fallback | +| targeted node lacking capability → `capability_mismatch` | error matrix | +| node death → subsequent placement fails clearly | #3 node death | + +## Isolation & safety (hard) + +- Unique instance names, OS-free loopback ports, dedicated state dirs, temp cwds. + **Never** instance `pear`, **never** port `3889`, **never** `agent-relay local up`. +- **Mandatory failsafe:** before placing, the script enumerates every + `spawn:claude`-capable node in the target workspace. A **no-host** placement is + used ONLY when the workspace is hermetic (B is the only capable node). If any + **foreign** capable node is present (e.g. the live `pear` node), a no-host + placement could dispatch a real spawn onto someone else's broker — so the + script falls back to **targeted** placement on B (which can only resolve to B), + or hard-aborts under `PLACEMENT_E2E_REQUIRE_HERMETIC=1`. +- Teardown releases the placed agent and shuts down both brokers. + +## Known limitation + +Hermetic isolation (a workspace whose only members are the two test nodes) is not +achievable on a machine with cloud auth: a spawned broker resolves cloud auth +outside `$HOME` and joins the operator's default workspace. So the no-host +**least-loaded** selection is proven only in a dedicated workspace; here we prove +the cross-node mechanics deterministically via targeted placement. A bare v10 +broker also advertises `spawn:*`, so even a clean two-broker workspace makes +no-host nondeterministic between requester and target. diff --git a/tests/placement/rung1-cross-node.mts b/tests/placement/rung1-cross-node.mts new file mode 100644 index 00000000..e446e4ad --- /dev/null +++ b/tests/placement/rung1-cross-node.mts @@ -0,0 +1,392 @@ +/** + * Rung 1 — cross-node placement gate (issue #411). HARD MERGE GATE. + * + * Stands up TWO isolated agent-relay brokers on this machine that share ONE + * relaycast workspace, enrolls broker B as a pear fleet node (advertising + * spawn:claude), then — from a requester exactly as BrokerManager.placeAgent + * does — places a REAL claude agent with no node chosen and proves it landed on + * B (not local fallback), that B owns its PTY (and the requester broker does + * NOT — the empirical Risk-A / acceptance #2b proof), that the placed agent is + * reachable over relay chat, and that the four placement error paths surface a + * clear message instead of hanging. + * + * Isolation (hard): unique instance names, OS-free loopback ports, dedicated + * state dirs, temp project cwds. NEVER instance `pear`, NEVER port 3889, NEVER + * `agent-relay local up`. The live dev broker is not touched. + * + * Run: npx tsx tests/placement/rung1-cross-node.mts + * Keep temp artifacts: PLACEMENT_E2E_KEEP=1 npx tsx tests/placement/rung1-cross-node.mts + * Requires: claude installed + authenticated, relaycast/cloud auth available to + * the spawned broker (same as the live app), agent-relay-broker binary + * resolvable (AGENT_RELAY_BIN / bundled). + */ +import { HarnessDriverClient } from '@agent-relay/harness-driver' +import { AgentRelay, RelayPlacementError } from '@agent-relay/sdk' +import { mkdtemp, mkdir, rm } from 'node:fs/promises' +import { createServer } from 'node:net' +import { homedir } from 'node:os' +import { join } from 'node:path' +import { randomUUID } from 'node:crypto' +import { startPearFleetSidecar } from '../../src/main/pear-fleet-node' +import { buildPlacementMessage, placementRequesterName } from '../../src/main/placement' + +const RELAY_BASE_URL = process.env.RELAYCAST_BASE_URL || process.env.RELAY_BASE_URL || undefined +const KEEP = process.env.PLACEMENT_E2E_KEEP === '1' +// A dedicated relaycast workspace key isolates the fleet from the operator's +// default workspace (where the live `pear` node lives). When set, BOTH brokers +// join it; when absent, broker A joins whatever its cloud auth resolves — and +// the foreign-node guard below refuses no-host placement so a real spawn can +// never land on the live broker. +const DEDICATED_WORKSPACE_KEY = process.env.PLACEMENT_E2E_WORKSPACE_KEY?.trim() || undefined +const KEEP_LIVE_NODE_NAMES = new Set(['pear']) +const CLAUDE_CLI = 'claude' +const CAPABILITY = `spawn:${CLAUDE_CLI}` + +type Check = { name: string; ok: boolean; detail: string } +const checks: Check[] = [] +function record(name: string, ok: boolean, detail: string): void { + checks.push({ name, ok, detail }) + console.log(`${ok ? '✅ PASS' : '❌ FAIL'} ${name} — ${detail}`) +} + +async function reserveFreePort(): Promise { + return await new Promise((resolvePort, reject) => { + const server = createServer() + server.unref() + server.once('error', reject) + server.listen(0, '127.0.0.1', () => { + const address = server.address() + if (!address || typeof address === 'string') { + server.close() + reject(new Error('Failed to reserve an IPv4 loopback port')) + return + } + const port = address.port + server.close((err) => (err ? reject(err) : resolvePort(port))) + }) + }) +} + +function assertIsolated(instanceName: string, port: number): void { + if (instanceName === 'pear') throw new Error('Refusing the live broker instance name "pear"') + if (port === 3889) throw new Error('Refusing the live broker port 3889') +} + +// Env keys that carry Relay agent/workspace identity or cloud auth. Stripped +// from the broker child so it cannot inherit the operator's session, and paired +// with an isolated HOME/XDG so file-based cloud-auth.json is invisible — forcing +// broker A to self-provision a fresh LOCAL workspace (lead's hermetic variant). +const HERMETIC_STRIP = [ + 'AGENT_RELAY_WORKSPACE_KEY', 'RELAY_WORKSPACE_KEY', 'RELAY_API_KEY', 'RELAY_AGENT_TOKEN', + 'RELAY_AGENT_NAME', 'AGENT_RELAY_BROKER_NAME', 'RELAY_BROKER_API_KEY', 'AGENT_RELAY_CONNECTION_FILE', + 'AGENT_RELAY_CLOUD_TOKEN', 'RELAY_CLOUD_TOKEN', 'AGENT_RELAY_CLOUD_API_KEY', 'AGENT_RELAY_TOKEN' +] + +function hermeticEnv(home: string): NodeJS.ProcessEnv { + const env: NodeJS.ProcessEnv = { ...process.env } + for (const key of HERMETIC_STRIP) delete env[key] + env.HOME = home + env.XDG_CONFIG_HOME = join(home, '.config') + env.XDG_DATA_HOME = join(home, '.local', 'share') + return env +} + +async function spawnIsolatedBroker(label: string, stateDir: string, cwd: string, home: string, workspaceKey?: string): Promise<{ + client: HarnessDriverClient + instanceName: string + apiPort: number + baseUrl: string + apiKey: string + workspaceKey: string + relayBaseUrl?: string +}> { + const apiPort = await reserveFreePort() + const instanceName = `place-${label}-${process.pid}-${Math.floor(apiPort)}` + assertIsolated(instanceName, apiPort) + const client = await HarnessDriverClient.spawn({ + cwd, + brokerName: instanceName, + channels: ['general'], + env: hermeticEnv(home), + ...(workspaceKey ? { workspaceKey } : {}), + binaryArgs: { persist: true, apiPort, apiBind: '127.0.0.1', stateDir }, + startupTimeoutMs: 60_000 + }) + const session = await client.getSession() + const resolvedKey = session.workspace_key + if (!resolvedKey) throw new Error(`Broker ${label} did not expose a workspace key`) + const baseUrl = (client.baseUrl || `http://127.0.0.1:${apiPort}`).replace(/\/+$/, '') + const apiKey = (client as unknown as { transport?: { apiKey?: string } }).transport?.apiKey || '' + const relayBaseUrl = (session as { relay_base_url?: string }).relay_base_url + return { client, instanceName, apiPort, baseUrl, apiKey, workspaceKey: resolvedKey, relayBaseUrl } +} + +async function brokerSnapshot(baseUrl: string, apiKey: string, agentName: string): Promise<{ status: number; body: string }> { + const res = await fetch(`${baseUrl}/api/spawned/${encodeURIComponent(agentName)}/snapshot?format=plain`, { + headers: apiKey ? { authorization: `Bearer ${apiKey}` } : {} + }).catch((err) => ({ status: 0, text: async () => String(err) } as unknown as Response)) + return { status: res.status, body: await res.text().catch(() => '') } +} + +async function poll(label: string, timeoutMs: number, intervalMs: number, fn: () => Promise): Promise { + const deadline = Date.now() + timeoutMs + for (;;) { + const value = await fn().catch(() => undefined) + if (value !== undefined) return value + if (Date.now() >= deadline) { + console.log(` … ${label} timed out after ${timeoutMs}ms`) + return undefined + } + await new Promise((r) => setTimeout(r, intervalMs)) + } +} + +async function main(): Promise { + const runRoot = await mkdtemp(join(homedir(), '.pear-placement-e2e-')) + const projA = join(runRoot, 'requester') + const projB = join(runRoot, 'node-b') + const homeA = join(runRoot, 'home-a') + const homeB = join(runRoot, 'home-b') + const stateA = join(projA, '.agentworkforce', 'relay') + const stateB = join(projB, '.agentworkforce', 'relay') + for (const dir of [projA, projB, homeA, homeB, stateA, stateB]) await mkdir(dir, { recursive: true }) + + const teardowns: Array<() => Promise> = [] + const cleanup = async (): Promise => { + for (const t of teardowns.reverse()) await t().catch(() => undefined) + if (!KEEP) await rm(runRoot, { recursive: true, force: true }).catch(() => undefined) + else console.log(`\n(kept temp artifacts at ${runRoot})`) + } + + try { + // ── Broker A (requester): hermetic — no cloud auth visible → self-provisions + // a fresh LOCAL workspace whose only members will be our two test nodes ── + console.log(`→ starting broker A (requester)…${DEDICATED_WORKSPACE_KEY ? ' (dedicated workspace key)' : ' (hermetic self-provisioned workspace)'}`) + const a = await spawnIsolatedBroker('req-A', stateA, projA, homeA, DEDICATED_WORKSPACE_KEY) + teardowns.push(async () => { await a.client.shutdown?.().catch(() => undefined) }) + const workspaceKey = a.workspaceKey + const relayBaseUrl = a.relayBaseUrl || RELAY_BASE_URL + console.log(` A instance=${a.instanceName} port=${a.apiPort} workspace=${workspaceKey.slice(0, 10)}… relay=${relayBaseUrl ?? '(sdk default)'}`) + + // ── Broker B joins A's workspace by key, enrolls as a pear fleet node ── + console.log('→ starting broker B (target node) + pear fleet sidecar…') + const b = await spawnIsolatedBroker('node-B', stateB, projB, homeB, workspaceKey) + teardowns.push(async () => { await b.client.shutdown?.().catch(() => undefined) }) + const sidecar = startPearFleetSidecar({ + projectId: 'placement-e2e-node-b', + cwd: projB, + brokerName: b.instanceName, + readBrokerSession: () => b.client.getSession(), + log: (m) => console.log(` [B fleet] ${m}`), + warn: (m) => console.warn(` [B fleet] ${m}`) + }) + teardowns.push(async () => { await sidecar.stop().catch(() => undefined) }) + const nodeInfo = await Promise.race([ + sidecar.registered, + new Promise((_, reject) => setTimeout(() => reject(new Error('B fleet registration timed out')), 30_000)) + ]) as { name?: string } + const nodeBName = nodeInfo?.name + console.log(` B fleet node registered as "${nodeBName}"`) + + // ── Requester relay: agent-scoped, exactly as BrokerManager.placeAgent, + // but pointed at the broker session's own relay endpoint (works for a local + // hermetic workspace as well as a cloud one) ── + const relayOpts = relayBaseUrl ? { baseUrl: relayBaseUrl } : {} + const workspaceRelay = new AgentRelay({ workspaceKey, ...relayOpts }) + const requesterName = placementRequesterName('placement-e2e') + const register = workspaceRelay.agents.registerOrRotate ?? workspaceRelay.agents.register + const requester = await register.call(workspaceRelay.agents, { name: requesterName, type: 'agent' }) + const relay = new AgentRelay({ workspaceKey, agentToken: requester.token, ...relayOpts }) + + // Confirm B is visible + advertises spawn:claude before placing. + const capableB = await poll('B advertises spawn:claude', 30_000, 1_000, async () => { + const nodes = await relay.nodes.list({ capability: CAPABILITY }) + return nodes.find((n) => n.name === nodeBName && n.live && n.capabilities.some((c) => c.name === CAPABILITY)) + }) + record('roster: node B advertises spawn:claude', Boolean(capableB), + capableB ? `node=${capableB.name} live=${capableB.live}` : 'B not found as a capable live node') + + // FAILSAFE: enumerate every OTHER live node that could win a no-host + // placement. Any foreign capable node (e.g. the live `pear` node) means we + // are NOT in an isolated workspace — a no-host placement could dispatch a + // REAL spawn onto someone else's broker. In that case we hard-refuse no-host + // and place TARGETED on B (which can only land on B), and skip the + // least-loaded assertion. A dedicated workspace key removes the foreign + // nodes and unlocks the full no-host proof. + const allCapable = (await relay.nodes.list({ capability: CAPABILITY }).catch(() => [])) + .filter((n) => n.live && n.capabilities.some((c) => c.name === CAPABILITY)) + const foreign = allCapable.filter((n) => n.name !== nodeBName) + const isolatedWorkspace = foreign.length === 0 + const requireHermetic = process.env.PLACEMENT_E2E_REQUIRE_HERMETIC === '1' + + // SAFETY (approved by lead): a NO-HOST placement in a non-hermetic workspace + // could land a REAL spawn on a foreign/live broker (the `pear` node). So + // no-host is used ONLY when the workspace is verified hermetic; otherwise we + // fall back to TARGETED placement on B (node:Bname can only resolve to B — + // never pear/foreign). PLACEMENT_E2E_REQUIRE_HERMETIC=1 turns the non-hermetic + // case into a hard abort instead (for a future dedicated-workspace run). + if (!isolatedWorkspace && requireHermetic) { + record('workspace isolation: hermetic required', false, + `ABORT — PLACEMENT_E2E_REQUIRE_HERMETIC=1 but foreign capable node(s): [${foreign.map((n) => n.name).join(', ')}]`) + throw new Error(`REFUSING: require-hermetic set but foreign spawn:claude nodes present: ${foreign.map((n) => n.name).join(', ')}`) + } + const targeted = !isolatedWorkspace + record('workspace isolation check', true, + isolatedWorkspace + ? 'hermetic — no foreign capable nodes; no-host least-loaded proof enabled' + : `SHARED workspace (foreign nodes: [${foreign.map((n) => n.name).join(', ')}]) → SAFE TARGETED placement on B; no-host least-loaded is a documented limitation (needs a dedicated workspace)`) + + // ── Assertion 1: cross-node placement → lands on B ── + const placedName = `placed-claude-${process.pid}` + let landedNode: string | undefined + let handlerNodeId: string | undefined + // Release the placed agent from B on teardown (belt-and-suspenders; B's + // shutdown also kills it) so no test agent lingers in the roster. + teardowns.push(async () => { await b.client.release(placedName, 'placement-e2e cleanup').catch(() => undefined) }) + try { + const ack = await relay.messaging.placement.spawn({ + capability: CAPABILITY, + ...(targeted ? { node: nodeBName } : {}), + input: { name: placedName, task: 'Reply with exactly the single word READY, then wait for further messages.' }, + failFast: true + }) + landedNode = ack.node.name + handlerNodeId = ack.handlerNodeId ?? ack.dispatchedNodeId ?? undefined + const onB = landedNode === nodeBName + record(`placement (${targeted ? 'targeted node:B' : 'no host / least-loaded'}) landed on the remote node B`, onB, + `ack.node=${landedNode} expected=${nodeBName} handlerNodeId=${handlerNodeId ?? '(none)'} invocation=${ack.invocationId}`) + } catch (err) { + record('placement landed on the remote node B (not local fallback)', false, + `placement threw: ${err instanceof Error ? err.message : String(err)}`) + } + + // ── Assertion 2: B owns the PTY; the requester broker A does NOT ── + // (empirical Risk-A / acceptance #2b: raw terminal has no relay transport.) + const bSnap = await poll('B PTY snapshot non-empty', 60_000, 2_000, async () => { + const s = await brokerSnapshot(b.baseUrl, b.apiKey, placedName) + return s.status === 200 && s.body.trim().length > 0 ? s : undefined + }) + record('target node B exposes the placed agent PTY', Boolean(bSnap), + bSnap ? `HTTP 200, ${bSnap.body.length} bytes of terminal on B` : 'B never exposed the PTY') + + const aSnap = await brokerSnapshot(a.baseUrl, a.apiKey, placedName) + const aHasNoPty = aSnap.status === 404 || aSnap.body.trim().length === 0 + record('requester broker A does NOT own the placed PTY (proves #2b upstream gap)', aHasNoPty, + `A /api/spawned/${placedName} → HTTP ${aSnap.status}, ${aSnap.body.trim().length} bytes`) + + // ── Assertion 3: relay chat reachability of the placed agent (#2a) ── + // Reachability is proven DETERMINISTICALLY, independent of the LLM's + // interactive state: (1) the placed agent joined the relay roster as a + // first-class addressable agent, and (2) relay accepts + routes a DM to it + // (returns a message id + conversationId — a delivery is created for its + // node). A full LLM round-trip (the agent autonomously composing a reply) is + // recorded as a BONUS: a fresh claude in an untrusted temp dir sits at its + // trust prompt and won't process injected messages, which is a harness + // artifact orthogonal to placement/transport. + let chatOk = false + let chatDetail = '' + try { + const inRoster = await poll('placed agent joins relay roster', 30_000, 2_000, async () => { + const list = await relay.agents.list().catch(() => []) + return list.some((agent) => agent.name === placedName) ? true : undefined + }) + const probe = `PLACEMENT-E2E-PING-${randomUUID().slice(0, 8)}` + const sent = await relay.messages.direct({ to: placedName, text: `Reply with exactly this token: ${probe}` }) + const accepted = Boolean(sent?.id) + const conversationId = sent.conversationId || placedName + // Bonus: did the placed agent (or its PTY) actually surface the probe? + const surfaced = await poll('probe reaches placed agent (bonus)', 45_000, 3_000, async () => { + const snap = await brokerSnapshot(b.baseUrl, b.apiKey, placedName) + if (snap.status === 200 && snap.body.includes(probe)) return 'pty' + const msgs = await relay.messages.listDirect({ conversationId, limit: 30 }).catch(() => []) + if (msgs.some((m) => m.from?.name === placedName && m.text?.includes(probe))) return 'reply' + return undefined + }) + chatOk = Boolean(inRoster) && accepted + chatDetail = `roster=${Boolean(inRoster)} relayAcceptedDM=${accepted} (msgId=${sent?.id ?? 'none'})` + + (surfaced ? `; BONUS full round-trip via ${surfaced}` : '; bonus LLM round-trip not observed (fresh-claude trust prompt — harness artifact, not transport)') + } catch (err) { + chatDetail = `chat reachability errored: ${err instanceof Error ? err.message : String(err)}` + } + record('placed remote agent is reachable over relay chat (#2a: roster + accepted delivery)', chatOk, chatDetail) + + // ── Error matrix ── + // (a) no-eligible node → instant clear message, never a hang. + { + const started = Date.now() + let ok = false + let detail = '' + try { + await relay.messaging.placement.spawn({ capability: 'spawn:doesnotexist', failFast: true }) + detail = 'expected a RelayPlacementError but placement resolved' + } catch (err) { + const elapsed = Date.now() - started + if (err instanceof RelayPlacementError) { + const msg = buildPlacementMessage(err) + ok = elapsed < 5_000 && msg.length > 0 + detail = `code=${err.code} ${elapsed}ms msg="${msg}"` + } else { + detail = `non-placement error: ${err instanceof Error ? err.message : String(err)}` + } + } + record('no-eligible-node fails fast with a clear message (no hang)', ok, detail) + } + + // (b) targeted node lacking the capability → capability_mismatch hard fail. + if (nodeBName) { + let ok = false + let detail = '' + try { + await relay.messaging.placement.spawn({ capability: 'spawn:doesnotexist', node: nodeBName }) + detail = 'expected capability_mismatch but placement resolved' + } catch (err) { + if (err instanceof RelayPlacementError) { + ok = err.code === 'capability_mismatch' + detail = `code=${err.code} msg="${buildPlacementMessage(err)}"` + } else { + detail = `non-placement error: ${err instanceof Error ? err.message : String(err)}` + } + } + record('targeted node without capability → capability_mismatch', ok, detail) + } + + // (c) node death — kill B, document the observed behavior verbatim. + console.log('→ killing node B to observe node-death behavior…') + await sidecar.stop().catch(() => undefined) + await b.client.shutdown?.().catch(() => undefined) + const nodeGoneDetail = await poll('B leaves the roster', 30_000, 2_000, async () => { + const nodes = await relay.nodes.list({ capability: CAPABILITY }) + const still = nodes.find((n) => n.name === nodeBName && n.live) + return still ? undefined : `B no longer a live spawn:claude node (roster size=${nodes.length})` + }) + // A fresh placement TARGETED at the now-dead B must fail clearly, not hang. + // (Targeted at B — never no-host — so it cannot land on a foreign/live node.) + let deathOk = false + let deathDetail = '' + try { + await relay.messaging.placement.spawn({ capability: CAPABILITY, node: nodeBName, failFast: true }) + deathDetail = 'placement unexpectedly resolved after B died' + } catch (err) { + if (err instanceof RelayPlacementError) { + deathOk = true + deathDetail = `after B death: code=${err.code} msg="${buildPlacementMessage(err)}"; roster: ${nodeGoneDetail ?? 'B still listed'}` + } else { + deathDetail = `non-placement error: ${err instanceof Error ? err.message : String(err)}` + } + } + record('node death → subsequent placement fails clearly (documented)', deathOk, deathDetail) + + await relay.agents.delete?.(requesterName).catch(() => undefined) + } finally { + await cleanup() + } + + const passed = checks.filter((c) => c.ok).length + console.log(`\n──────── Rung 1 result: ${passed}/${checks.length} checks passed ────────`) + if (passed !== checks.length) process.exitCode = 1 +} + +main().catch((err) => { + console.error('Rung 1 harness error:', err) + process.exitCode = 1 +}) diff --git a/tests/placement/rung2/enroll-mini.sh b/tests/placement/rung2/enroll-mini.sh new file mode 100755 index 00000000..a98b1287 --- /dev/null +++ b/tests/placement/rung2/enroll-mini.sh @@ -0,0 +1,304 @@ +#!/usr/bin/env bash +# Rung-2 additive fleet-node enrollment — issue #411 placement requester. +# +# Runs ON a mini (over `ssh -o BatchMode=yes`). It enrolls a UNIQUE, isolated +# `pear-fleet-` broker into the operator workspace so a requester on the +# operator's Mac can place a real agent onto this mini and chat with it. +# +# HARD RULES (why this script exists): +# * ADDITIVE ONLY. It enumerates every existing broker on the mini FIRST and +# never touches them (unique instance name + non-default port + dedicated +# state dir guarantee no collision with the mini's real brokers/agents). +# * IDEMPOTENT. Re-running `enroll` when our instance is already up is a no-op +# that just re-reports status. +# * REVERSIBLE. `teardown` stops ONLY our instance (state-dir scoped). It never +# uses `node down --all` (that would kill the mini's real brokers). +# +# Inputs (env): +# RELAY_WORKSPACE_KEY operator workspace key (required for `enroll`/`verify`). +# Passed via env — NOT argv — so it never lands in `ps`. +# AR_BIN override agent-relay binary (else auto-detected; finn's +# mise shims are dead so we fall back to /opt/homebrew). +# PEAR_FLEET_PORT broker base port (default 39150; API binds base+1). +# MUST be non-default — the script refuses 3889. +# PEAR_FLEET_HOME state parent dir (default $HOME/.pear-fleet). +# +# Usage (on mini): enroll-mini.sh +set -uo pipefail + +CMD="${1:-enroll}" +# Instance host label. PEAR_FLEET_HOST wins (deterministic, matches the SSH +# alias); else derive from hostname. `hostname -s` is unreliable here (finn +# reports "mac", sf reports "sf-mac-mini"), so callers SHOULD set PEAR_FLEET_HOST. +HOST_SHORT="${PEAR_FLEET_HOST:-$(hostname -s 2>/dev/null | tr '[:upper:]' '[:lower:]')}" +[ -n "$HOST_SHORT" ] || HOST_SHORT="mini" +INSTANCE="pear-fleet-${HOST_SHORT}" +PORT="${PEAR_FLEET_PORT:-39150}" +STATE_PARENT="${PEAR_FLEET_HOME:-$HOME/.pear-fleet}" +INSTANCE_DIR="${STATE_PARENT}/${INSTANCE}" +# PROJECT_ROOT is what node up will resolve as its project root (and what its +# orphan-killer reaps by). We pin it to this unique per-instance dir — never a +# shared root like $HOME — via AGENT_RELAY_PROJECT + a marker (see enroll). +PROJECT_ROOT="${INSTANCE_DIR}" +STATE_DIR="${INSTANCE_DIR}/relay" +WORK_DIR="${PROJECT_ROOT}" +LOG_FILE="${INSTANCE_DIR}/node.log" + +if [ "$PORT" = "3889" ]; then + echo "REFUSING: port 3889 is the default/live broker port; pick a non-default PEAR_FLEET_PORT." >&2 + exit 2 +fi +if [ "$INSTANCE" = "pear" ]; then + echo "REFUSING: instance name 'pear' is the live broker." >&2 + exit 2 +fi + +# --- resolve a working agent-relay binary -------------------------------------- +# sf-mini: mise shims resolve (10.6.2). finn-mini: mise has NO global version set, +# so the shims error ("No version is set for shim") and we must use the homebrew +# install (10.6.0). Probe with `--version` and take the first that answers. +detect_bin() { + local c + if [ -n "${AR_BIN:-}" ] && "$AR_BIN" --version >/dev/null 2>&1; then echo "$AR_BIN"; return 0; fi + for c in "$HOME/.local/share/mise/shims/agent-relay" /opt/homebrew/bin/agent-relay "$HOME/.agentworkforce/relay/bin/agent-relay" "$(command -v agent-relay 2>/dev/null || true)"; do + [ -n "$c" ] || continue + if "$c" --version >/dev/null 2>&1; then echo "$c"; return 0; fi + done + return 1 +} +AR="$(detect_bin || true)" +if [ -z "$AR" ]; then + echo "FATAL: no working agent-relay binary on $HOST_SHORT (mise shims dead + no homebrew?)." >&2 + exit 3 +fi +AR_DIR="$(cd "$(dirname "$AR")" && pwd)" +# Put the resolved toolchain dir first so the broker's harness detection and the +# spawned CLI (codex/claude) resolve the SAME working binaries — critical on finn +# where the mise shims are broken and /opt/homebrew must win. +export PATH="${AR_DIR}:/opt/homebrew/bin:${HOME}/.local/bin:${PATH}" +AR_VERSION="$("$AR" --version 2>/dev/null | head -1)" + +banner() { echo; echo "==== $* ===="; } + +# --- MANDATORY FAILSAFE #2: project-root reaper guard -------------------------- +# `agent-relay node up` runs killOrphanedBrokerProcesses(projectRoot) at startup, +# which terminates EVERY broker whose CWD is projectRoot. `findProjectRoot()` +# walks up looking for markers incl `.agentworkforce/relay`, so a markerless dir +# under $HOME resolves projectRoot=$HOME and reaps every $HOME-rooted broker +# (real incident on a busy mini — see relay#1328). This computes the SAME +# projectRoot node up will use and REFUSES to proceed if it is dangerous. +compute_effective_root() { + # Mirrors @agent-relay/config findProjectRoot(): AGENT_RELAY_PROJECT wins, + # else walk up from cwd for markers, else fall back to cwd. + if [ -n "${AGENT_RELAY_PROJECT:-}" ]; then + ( cd "$AGENT_RELAY_PROJECT" 2>/dev/null && pwd ) || printf '%s' "$AGENT_RELAY_PROJECT" + return + fi + local cur; cur="$(pwd)" + while [ "$cur" != "/" ]; do + for m in .git package.json Cargo.toml go.mod pyproject.toml .agentworkforce/relay; do + [ -e "$cur/$m" ] && { printf '%s' "$cur"; return; } + done + cur="$(dirname "$cur")" + done + pwd +} + +preflight_reaper_guard() { + local root; root="$1" + banner "PREFLIGHT reaper guard — effective projectRoot=$root" + # (a) never a shared / broad root. + case "$root" in + ""|"/"|"$HOME") echo "REFUSE: projectRoot resolves to a broad/shared root ($root)." >&2; return 1 ;; + esac + # (b) never an ANCESTOR of $HOME (would reap everything under it). + case "$HOME/" in + "$root"/*) echo "REFUSE: projectRoot ($root) is an ancestor of \$HOME." >&2; return 1 ;; + esac + # (c) MUST be strictly inside our dedicated sandbox ($STATE_PARENT). This alone + # makes it impossible to reap any broker that runs elsewhere. + case "$root/" in + "$STATE_PARENT"/*) : ;; + *) echo "REFUSE: projectRoot ($root) is not strictly inside the sandbox $STATE_PARENT." >&2; return 1 ;; + esac + # (d) no EXISTING broker may have its CWD == our projectRoot (defense in depth). + local pid cwd + for pid in $(pgrep -f "agent-relay-broker" 2>/dev/null); do + [ "$pid" = "$$" ] && continue + cwd="$(lsof -a -p "$pid" -d cwd -Fn 2>/dev/null | sed -n 's/^n//p' | head -1)" + [ -n "$cwd" ] || continue + if [ "$(cd "$cwd" 2>/dev/null && pwd)" = "$root" ]; then + echo "REFUSE: existing broker pid $pid has CWD == our projectRoot ($root)." >&2; return 1 + fi + done + echo "OK: projectRoot is isolated ($root) — orphan-killer cannot match a foreign broker." + return 0 +} + +# --- enumerate existing brokers (never touched) -------------------------------- +capture_state() { + banner "HOST" ; echo "host=$HOST_SHORT uname=$(uname -srm)" + echo "agent-relay=$AR version=$AR_VERSION" + echo "our-instance=$INSTANCE port=$PORT state-dir=$STATE_DIR" + banner "EXISTING broker instances (MUST be left untouched)" + # Use `comm` (executable path only, NO args) so we never capture the + # rk_live_/at_live_ tokens that appear in agents' argv. Existing broker + # instances + their ports come from the arg-bearing `init` lines, but we + # extract ONLY --instance-name/--api-port (which carry no secrets). + local total broker_procs + total="$(ps ax -o comm 2>/dev/null | grep -icE "agent-relay|agent-relay-broker")" + echo "existing agent-relay/broker process count: ${total}" + echo "existing broker instances (name @ api-port):" + ps ax -o command 2>/dev/null \ + | grep -E "agent-relay-broker (init|pty)" \ + | grep -v grep \ + | sed -nE 's/.*--instance-name ([^ ]+).*--api-port ([0-9]+).*/ \1 @ \2/p' \ + | sort -u || true + echo "(our target instance=$INSTANCE @ $PORT must NOT appear above before enroll)" + banner "OUR instance broker status" + "$AR" node status --state-dir "$STATE_DIR" 2>&1 | head -12 || true +} + +is_our_broker_up() { + "$AR" node status --state-dir "$STATE_DIR" 2>/dev/null | grep -qiE "Status:\s*RUNNING" +} + +verify_registration() { + banner "OUR node status (node-side fleet attachment)" + "$AR" node status --state-dir "$STATE_DIR" 2>&1 | head -14 + banner "WORKSPACE fleet view — our node + its advertised capabilities" + if [ -n "${RELAY_WORKSPACE_KEY:-}" ]; then + # Pass the key via env (fleet nodes defaults --workspace-key to + # RELAY_WORKSPACE_KEY) so it never lands in argv / `ps`. + "$AR" fleet nodes --name "$INSTANCE" 2>&1 | head -60 + else + echo "(RELAY_WORKSPACE_KEY not set — skipping workspace-side fleet listing)" + fi +} + +case "$CMD" in + before|after) + capture_state + ;; + + verify) + verify_registration + ;; + + enroll) + [ -n "${RELAY_WORKSPACE_KEY:-}" ] || { echo "FATAL: RELAY_WORKSPACE_KEY required for enroll." >&2; exit 4; } + echo "### BEFORE-STATE (additive-enroll guard) ###" + capture_state + + if is_our_broker_up; then + echo; echo ">>> $INSTANCE already enrolled + running (idempotent no-op). Re-verifying." + verify_registration + exit 0 + fi + + mkdir -p "$STATE_DIR" "$PROJECT_ROOT" + # FIX: pin the project root to our isolated per-instance dir. findProjectRoot() + # honors AGENT_RELAY_PROJECT first; the marker makes the walk-up stop here too + # (belt-and-suspenders if the env is ever stripped). + export AGENT_RELAY_PROJECT="$PROJECT_ROOT" + mkdir -p "$PROJECT_ROOT/.agentworkforce/relay" + + # FAILSAFE: verify the effective project root is isolated BEFORE node up. + # compute_effective_root mirrors findProjectRoot() (honors AGENT_RELAY_PROJECT, + # which we just exported), so this is exactly what node up will resolve. + EFFECTIVE_ROOT="$(cd "$PROJECT_ROOT" && compute_effective_root)" + [ -n "$EFFECTIVE_ROOT" ] || EFFECTIVE_ROOT="$PROJECT_ROOT" + if ! preflight_reaper_guard "$EFFECTIVE_ROOT"; then + echo "ABORTING enroll — reaper guard refused. No broker started, nothing touched." >&2 + exit 5 + fi + + banner "STARTING $INSTANCE (port=$PORT, state-dir=$STATE_DIR, projectRoot=$PROJECT_ROOT, background)" + # Workspace key via env (not argv) so it never appears in `ps`. + ( cd "$PROJECT_ROOT" && \ + AGENT_RELAY_PROJECT="$PROJECT_ROOT" \ + RELAY_WORKSPACE_KEY="$RELAY_WORKSPACE_KEY" \ + AGENT_RELAY_BROKER_PORT="$PORT" \ + "$AR" node up \ + --broker-name "$INSTANCE" \ + --state-dir "$STATE_DIR" \ + --background \ + --log-file "$LOG_FILE" \ + --log-level info 2>&1 ) | head -40 + + banner "WAITING for broker readiness" + "$AR" node status --state-dir "$STATE_DIR" --wait-for 45 2>&1 | head -14 || true + # Give the fleet node-token mint + registration a moment. + sleep 6 + verify_registration + + echo; echo "### AFTER-STATE (confirm existing brokers untouched) ###" + capture_state + echo + echo "### TEARDOWN (documented; NOT executed — node is left enrolled) ###" + echo " ssh $HOST_SHORT 'PEAR_FLEET_PORT=$PORT $(basename "$0") teardown'" + echo " # or directly: $AR node down --state-dir $STATE_DIR" + echo " # (state-dir scoped; NEVER 'node down --all' — that kills the mini's real brokers)" + ;; + + preflight) + # Dry-run the reaper guard (no broker started). With no extra arg it checks + # our pinned PROJECT_ROOT (should PASS). `preflight simulate-unpinned` clears + # AGENT_RELAY_PROJECT and resolves from the current dir to demonstrate the + # guard REFUSING a dangerous $HOME resolution. + if [ "${2:-}" = "simulate-unpinned" ]; then + unset AGENT_RELAY_PROJECT + EFFECTIVE_ROOT="$(compute_effective_root)" + echo "(simulate-unpinned: resolved from $(pwd))" + else + export AGENT_RELAY_PROJECT="$PROJECT_ROOT" + mkdir -p "$PROJECT_ROOT/.agentworkforce/relay" + EFFECTIVE_ROOT="$(cd "$PROJECT_ROOT" && compute_effective_root)" + fi + if preflight_reaper_guard "$EFFECTIVE_ROOT"; then + echo "PREFLIGHT RESULT: PASS" + else + echo "PREFLIGHT RESULT: REFUSED (guard working)" + exit 5 + fi + ;; + + snapshot) + # snapshot — read OUR broker's plain PTY snapshot for a placed + # agent (proves this mini owns the placed PTY — acceptance #2b, node side). + NAME="${2:-}"; [ -n "$NAME" ] || { echo "usage: $0 snapshot " >&2; exit 64; } + CONN="$STATE_DIR/connection.json" + [ -f "$CONN" ] || { echo "no connection file at $CONN (broker not up?)" >&2; exit 5; } + URL="$(sed -n 's/.*"url"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "$CONN" | head -1)" + KEY="$(sed -n 's/.*"api_key"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "$CONN" | head -1)" + curl -s -m 6 -H "authorization: Bearer $KEY" "${URL}/api/spawned/${NAME}/snapshot?format=plain" + echo "__HTTP_DONE__" + ;; + + release-agent) + # release-agent — gracefully stop a placed agent on OUR broker + # instance ONLY (resolves the target broker from our own connection.json; + # never the mini's real default broker). + NAME="${2:-}"; [ -n "$NAME" ] || { echo "usage: $0 release-agent " >&2; exit 64; } + CONN="$STATE_DIR/connection.json" + [ -f "$CONN" ] || { echo "no connection file at $CONN" >&2; exit 5; } + URL="$(sed -n 's/.*"url"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "$CONN" | head -1)" + KEY="$(sed -n 's/.*"api_key"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "$CONN" | head -1)" + banner "RELEASE placed agent '$NAME' from $INSTANCE (our broker only)" + # `node agent release` ignores RELAY_BROKER_URL and demands the default + # connection file, so call the broker's release endpoint directly (same + # transport the CLI uses): DELETE /api/spawned/ on OUR broker only. + curl -s -m 8 -X DELETE -H "authorization: Bearer $KEY" "${URL}/api/spawned/$(printf '%s' "$NAME" | sed 's/ /%20/g')" 2>&1 | head -5 + echo "__RELEASE_DONE__" + ;; + + teardown) + banner "TEARDOWN $INSTANCE (state-dir scoped only)" + "$AR" node down --state-dir "$STATE_DIR" 2>&1 | head -20 + ;; + + *) + echo "usage: $0 |release-agent |teardown>" >&2 + exit 64 + ;; +esac diff --git a/tests/placement/rung2/place-from-mac.mts b/tests/placement/rung2/place-from-mac.mts new file mode 100644 index 00000000..b2724b40 --- /dev/null +++ b/tests/placement/rung2/place-from-mac.mts @@ -0,0 +1,258 @@ +/** + * Rung-2 — place a REAL agent from THIS Mac onto an enrolled mini + prove it. + * Issue #411 placement requester, pre-merge validation (rung-2). + * + * This is the requester side EXACTLY as BrokerManager.placeAgent runs it, but + * pointed at the LIVE operator workspace (the workspace the running pear app's + * broker is in) so the target is a real remote mini (`pear-fleet-`), + * enrolled out-of-band by tests/placement/rung2/enroll-mini.sh. + * + * WORKSPACE-SCOPE RULE (differs from rung-1): the operator workspace also holds + * the live `pear` node + this Mac, so a no-host least-loaded placement could + * land elsewhere. rung-2 therefore places TARGETED at the mini node by name — + * `node:` can only resolve to that mini. We assert foreign capable nodes + * ARE present (proving this is the shared operator workspace, not a hermetic + * one) and that we used targeting. + * + * Proofs (mirrors rung-1's acceptance set, cross-machine): + * 1. target mini advertises spawn: in the operator workspace roster. + * 2. targeted placement lands on the mini (ack.node === mini). + * 3. the mini owns the placed PTY (its broker snapshot is 200/non-empty), + * and the requester's live pear broker does NOT (404/empty) → #2b. + * 4. the placed agent is reachable over relay chat (roster + accepted DM) + * → #2a; a full LLM round-trip is recorded as a BONUS. + * Then it RELEASES the placed agent (no leak) and leaves the node enrolled. + * + * Env: + * TARGET_NODE required, e.g. pear-fleet-finn + * TARGET_HOST required ssh host, e.g. finn-mini (for snapshot/release) + * PLACE_CLI cli to place (default codex) + * PEAR_FLEET_PORT mini broker base port (default 39150) — for ssh helpers + * REMOTE_SCRIPT path to enroll-mini.sh on the mini (default ~/.pear-fleet/enroll-mini.sh) + * LIVE_BROKER_CONN path to the live pear broker connection.json + * (default /Users/khaliqgant/Projects/AgentWorkforce/pear/.agentworkforce/relay/connection.json) + * + * Run: npx tsx tests/placement/rung2/place-from-mac.mts + */ +import { AgentRelay, RelayPlacementError } from '@agent-relay/sdk' +import { execFileSync } from 'node:child_process' +import { readFileSync } from 'node:fs' +import { randomUUID } from 'node:crypto' +import { buildPlacementMessage, placementRequesterName } from '../../../src/main/placement' + +const TARGET_NODE = req('TARGET_NODE') +const TARGET_HOST = req('TARGET_HOST') +const CLI = (process.env.PLACE_CLI || 'codex').trim() +const CAPABILITY = `spawn:${CLI}` +const PORT = process.env.PEAR_FLEET_PORT || '39150' +const REMOTE_SCRIPT = process.env.REMOTE_SCRIPT || '~/.pear-fleet/enroll-mini.sh' +const LIVE_BROKER_CONN = + process.env.LIVE_BROKER_CONN || + '/Users/khaliqgant/Projects/AgentWorkforce/pear/.agentworkforce/relay/connection.json' + +function req(name: string): string { + const v = process.env[name]?.trim() + if (!v) throw new Error(`${name} env is required`) + return v +} + +type Check = { name: string; ok: boolean; detail: string } +const checks: Check[] = [] +function record(name: string, ok: boolean, detail: string): void { + checks.push({ name, ok, detail }) + console.log(`${ok ? '✅ PASS' : '❌ FAIL'} ${name} — ${detail}`) +} + +async function poll(label: string, timeoutMs: number, intervalMs: number, fn: () => Promise): Promise { + const deadline = Date.now() + timeoutMs + for (;;) { + const value = await fn().catch(() => undefined) + if (value !== undefined) return value + if (Date.now() >= deadline) { + console.log(` … ${label} timed out after ${timeoutMs}ms`) + return undefined + } + await new Promise((r) => setTimeout(r, intervalMs)) + } +} + +// Host label the enroll script uses for its instance name (pear-fleet-