mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-06 06:41:56 +00:00
feat(cli): ping-peer CLI + MCP ping_peer for peer messaging (#1195)
* feat(cli): add ping-peer CLI and MCP ping_peer for peer messaging Promote resume-if-inactive + wait-active + POST message into a first-class CLI command and session MCP tool so agents stop reinventing JWT+curl. Fixes #1194 Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cli): do not auto-approve MCP ping_peer Cross-session messaging can resume a peer and inject a prompt, so keep permission-mode gating (Codex PR review Major on #1195). Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cli): require approval for ping_peer in read-only mode Read-only auto-approve treated non-write names as safe; ping_peer can still resume a peer and inject prompts, so gate it like a write tool. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cli): keep ping_peer out of Claude --allowedTools toolNames still registers the MCP tool, but Claude auto-allow must not pre-approve cross-session resume+inject without a permission prompt. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cli): re-check session active before ping-peer send List/get can race; POST /messages still 409s if the target flips inactive before send. Resume+wait again (and re-gate pi) immediately before POST. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
HeavyGee
@@ -36,3 +36,28 @@ describe('resolveToolAutoApprovalDecision skill_lookup', () => {
|
||||
)).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
describe('resolveToolAutoApprovalDecision ping_peer', () => {
|
||||
it.each([
|
||||
'ping_peer',
|
||||
'mcp__hapi__ping_peer',
|
||||
'hapi_ping_peer',
|
||||
'Ping Peer Session'
|
||||
])('does not auto-approve %s in default mode', (toolName) => {
|
||||
expect(resolveToolAutoApprovalDecision('default', toolName, 'call-1')).toBeNull()
|
||||
})
|
||||
|
||||
it.each([
|
||||
'ping_peer',
|
||||
'mcp__hapi__ping_peer',
|
||||
'hapi_ping_peer',
|
||||
'Ping Peer Session'
|
||||
])('does not auto-approve %s in read-only mode', (toolName) => {
|
||||
expect(resolveToolAutoApprovalDecision('read-only', toolName, 'call-1')).toBeNull()
|
||||
})
|
||||
|
||||
it('still auto-approves unrelated read tools in read-only mode', () => {
|
||||
expect(resolveToolAutoApprovalDecision('read-only', 'Read', 'call-1')).toBe('approved')
|
||||
expect(resolveToolAutoApprovalDecision('read-only', 'grep', 'call-2')).toBe('approved')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -32,8 +32,21 @@ const AUTO_APPROVE_EXACT_TOOL_NAMES = new Set([
|
||||
'happy__skill_lookup',
|
||||
'mcp__hapi__skill_lookup'
|
||||
]);
|
||||
// ping_peer intentionally omitted from always-approve: it can resume another
|
||||
// session and inject a prompt into a peer, so permission modes must still gate
|
||||
// it (Codex PR #1195). Treat it as write-like in read-only so ACP titles such as
|
||||
// "Ping Peer Session" also require approval.
|
||||
const AUTO_APPROVE_TOOL_ID_HINTS = ['change_title', 'save_memory'];
|
||||
const AUTO_APPROVE_WRITE_TOOL_HINTS = ['write', 'edit', 'create', 'delete', 'patch', 'fs-edit'];
|
||||
const SENSITIVE_TOOL_NAME_HINTS = ['ping_peer', 'ping peer'];
|
||||
const AUTO_APPROVE_WRITE_TOOL_HINTS = [
|
||||
'write',
|
||||
'edit',
|
||||
'create',
|
||||
'delete',
|
||||
'patch',
|
||||
'fs-edit',
|
||||
...SENSITIVE_TOOL_NAME_HINTS
|
||||
];
|
||||
|
||||
export function resolveToolAutoApprovalDecision(
|
||||
mode: PermissionMode | undefined,
|
||||
|
||||
@@ -0,0 +1,399 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
PingPeerError,
|
||||
exitCodeForPingPeerError,
|
||||
pingPeer,
|
||||
resolveSessionByPrefix,
|
||||
type PingPeerSessionSummary
|
||||
} from './pingPeer'
|
||||
|
||||
type MockResponse = {
|
||||
status: number
|
||||
data: unknown
|
||||
}
|
||||
|
||||
function createHttpMock(handlers: {
|
||||
post?: (url: string, body?: unknown) => MockResponse | Promise<MockResponse>
|
||||
get?: (url: string) => MockResponse | Promise<MockResponse>
|
||||
}) {
|
||||
return {
|
||||
post: vi.fn(async (url: string, body?: unknown) => {
|
||||
if (!handlers.post) {
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
}
|
||||
return handlers.post(url, body)
|
||||
}),
|
||||
get: vi.fn(async (url: string) => {
|
||||
if (!handlers.get) {
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
return handlers.get(url)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
describe('resolveSessionByPrefix', () => {
|
||||
const sessions: PingPeerSessionSummary[] = [
|
||||
{ id: 'aaaaaaaa-1111-1111-1111-111111111111', active: true, metadata: { name: 'A' } },
|
||||
{ id: 'aaaaaaab-2222-2222-2222-222222222222', active: false, metadata: { name: 'B' } },
|
||||
{ id: 'bbbbbbbb-3333-3333-3333-333333333333', active: true, metadata: { name: 'C' } }
|
||||
]
|
||||
|
||||
it('resolves a unique id prefix', () => {
|
||||
expect(resolveSessionByPrefix(sessions, 'bbbb').id).toBe(sessions[2]!.id)
|
||||
})
|
||||
|
||||
it('prefers an exact id match', () => {
|
||||
expect(resolveSessionByPrefix(sessions, sessions[0]!.id).id).toBe(sessions[0]!.id)
|
||||
})
|
||||
|
||||
it('refuses ambiguous prefixes', () => {
|
||||
expect(() => resolveSessionByPrefix(sessions, 'aaaa')).toThrow(PingPeerError)
|
||||
try {
|
||||
resolveSessionByPrefix(sessions, 'aaaa')
|
||||
} catch (error) {
|
||||
expect(error).toBeInstanceOf(PingPeerError)
|
||||
expect((error as PingPeerError).code).toBe('ambiguous')
|
||||
}
|
||||
})
|
||||
|
||||
it('refuses unknown prefixes', () => {
|
||||
expect(() => resolveSessionByPrefix(sessions, 'zzzz')).toThrowError(/no session matching/)
|
||||
})
|
||||
})
|
||||
|
||||
describe('pingPeer', () => {
|
||||
let nowMs: number
|
||||
let sleepCalls: number[]
|
||||
|
||||
beforeEach(() => {
|
||||
nowMs = 1_000_000
|
||||
sleepCalls = []
|
||||
})
|
||||
|
||||
it('sends to an already-active session without resume', async () => {
|
||||
const sessionId = '05d9f0f2-9273-4137-933c-07459a1146a2'
|
||||
const http = createHttpMock({
|
||||
post: (url, body) => {
|
||||
if (url.endsWith('/api/auth')) {
|
||||
expect(body).toEqual({ accessToken: 'tok' })
|
||||
return { status: 200, data: { token: 'jwt' } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
|
||||
expect(body).toEqual({ text: 'hello peer' })
|
||||
return { status: 200, data: { ok: true } }
|
||||
}
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
},
|
||||
get: (url) => {
|
||||
if (url.endsWith('/api/sessions')) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
sessions: [{
|
||||
id: sessionId,
|
||||
active: true,
|
||||
metadata: { name: 'Orchestrator', flavor: 'cursor' }
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}`)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
session: {
|
||||
id: sessionId,
|
||||
active: true,
|
||||
metadata: { name: 'Orchestrator', flavor: 'cursor' }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
})
|
||||
|
||||
const result = await pingPeer({
|
||||
sessionIdPrefix: '05d9f0f2',
|
||||
message: 'hello peer',
|
||||
accessToken: 'tok',
|
||||
apiUrl: 'http://127.0.0.1:3006',
|
||||
http: http as never
|
||||
})
|
||||
|
||||
expect(result).toEqual({
|
||||
sessionId,
|
||||
name: 'Orchestrator',
|
||||
resumed: false
|
||||
})
|
||||
expect(http.post).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('resumes an inactive session, waits for active, then sends', async () => {
|
||||
const sessionId = 'aaaaaaaa-1111-1111-1111-111111111111'
|
||||
let active = false
|
||||
let polls = 0
|
||||
|
||||
const http = createHttpMock({
|
||||
post: (url) => {
|
||||
if (url.endsWith('/api/auth')) {
|
||||
return { status: 200, data: { token: 'jwt' } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/resume`)) {
|
||||
return { status: 200, data: { type: 'success', sessionId, resumed: true } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
|
||||
expect(active).toBe(true)
|
||||
return { status: 200, data: { ok: true } }
|
||||
}
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
},
|
||||
get: (url) => {
|
||||
if (url.endsWith('/api/sessions') && !url.includes(sessionId)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
sessions: [{
|
||||
id: sessionId,
|
||||
active: false,
|
||||
metadata: { name: 'Peer', flavor: 'claude' }
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}`)) {
|
||||
polls += 1
|
||||
if (polls >= 3) {
|
||||
active = true
|
||||
}
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
session: {
|
||||
id: sessionId,
|
||||
active,
|
||||
metadata: { name: 'Peer', flavor: 'claude' }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
})
|
||||
|
||||
const result = await pingPeer({
|
||||
sessionIdPrefix: 'aaaaaaaa',
|
||||
message: 'wake up',
|
||||
accessToken: 'tok',
|
||||
apiUrl: 'http://hub.test',
|
||||
waitActiveSecs: 10,
|
||||
http: http as never,
|
||||
now: () => nowMs,
|
||||
sleep: async (ms) => {
|
||||
sleepCalls.push(ms)
|
||||
nowMs += ms
|
||||
}
|
||||
})
|
||||
|
||||
expect(result.resumed).toBe(true)
|
||||
expect(result.sessionId).toBe(sessionId)
|
||||
expect(sleepCalls.length).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('re-checks active before send when the list snapshot was stale', async () => {
|
||||
const sessionId = 'bbbbbbbb-1111-1111-1111-111111111111'
|
||||
let active = false
|
||||
let resumeCalls = 0
|
||||
|
||||
const http = createHttpMock({
|
||||
post: (url) => {
|
||||
if (url.endsWith('/api/auth')) {
|
||||
return { status: 200, data: { token: 'jwt' } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/resume`)) {
|
||||
resumeCalls += 1
|
||||
return { status: 200, data: { type: 'success', sessionId, resumed: true } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
|
||||
expect(active).toBe(true)
|
||||
expect(resumeCalls).toBe(1)
|
||||
return { status: 200, data: { ok: true } }
|
||||
}
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
},
|
||||
get: (url) => {
|
||||
if (url.endsWith('/api/sessions') && !url.includes(sessionId)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
sessions: [{
|
||||
id: sessionId,
|
||||
// Stale snapshot: list claims active, live GET disagrees.
|
||||
active: true,
|
||||
metadata: { name: 'Stale', flavor: 'claude' }
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}`)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
session: {
|
||||
id: sessionId,
|
||||
active,
|
||||
metadata: { name: 'Stale', flavor: 'claude' }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
})
|
||||
|
||||
const result = await pingPeer({
|
||||
sessionIdPrefix: 'bbbbbbbb',
|
||||
message: 'still here?',
|
||||
accessToken: 'tok',
|
||||
apiUrl: 'http://hub.test',
|
||||
waitActiveSecs: 10,
|
||||
http: http as never,
|
||||
now: () => nowMs,
|
||||
sleep: async (ms) => {
|
||||
sleepCalls.push(ms)
|
||||
nowMs += ms
|
||||
// Become active only after resume has been requested.
|
||||
if (resumeCalls > 0) {
|
||||
active = true
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
expect(result.resumed).toBe(true)
|
||||
expect(resumeCalls).toBe(1)
|
||||
})
|
||||
|
||||
it('waits for piSessionId before sending to a pi session', async () => {
|
||||
const sessionId = 'piiiiiii-1111-1111-1111-111111111111'
|
||||
let piSessionId: string | undefined
|
||||
let getCount = 0
|
||||
|
||||
const http = createHttpMock({
|
||||
post: (url) => {
|
||||
if (url.endsWith('/api/auth')) {
|
||||
return { status: 200, data: { token: 'jwt' } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
|
||||
expect(piSessionId).toBe('pi-ready-1')
|
||||
return { status: 200, data: { ok: true } }
|
||||
}
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
},
|
||||
get: (url) => {
|
||||
if (url.endsWith('/api/sessions') && !url.includes(sessionId)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
sessions: [{
|
||||
id: sessionId,
|
||||
active: true,
|
||||
metadata: { name: 'Pi', flavor: 'pi' }
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}`)) {
|
||||
getCount += 1
|
||||
if (getCount >= 2) {
|
||||
piSessionId = 'pi-ready-1'
|
||||
}
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
session: {
|
||||
id: sessionId,
|
||||
active: true,
|
||||
metadata: { name: 'Pi', flavor: 'pi', piSessionId }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
})
|
||||
|
||||
await pingPeer({
|
||||
sessionIdPrefix: 'piiiiiii',
|
||||
message: 'hi pi',
|
||||
accessToken: 'tok',
|
||||
apiUrl: 'http://hub.test',
|
||||
waitActiveSecs: 5,
|
||||
http: http as never,
|
||||
now: () => nowMs,
|
||||
sleep: async (ms) => {
|
||||
nowMs += ms
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('maps resume failures to resume_failed', async () => {
|
||||
const sessionId = 'deadbeef-1111-1111-1111-111111111111'
|
||||
const http = createHttpMock({
|
||||
post: (url) => {
|
||||
if (url.endsWith('/api/auth')) {
|
||||
return { status: 200, data: { token: 'jwt' } }
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}/resume`)) {
|
||||
return {
|
||||
status: 503,
|
||||
data: { type: 'error', code: 'no_machine_online', message: 'no runner' }
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected POST ${url}`)
|
||||
},
|
||||
get: (url) => {
|
||||
if (url.endsWith('/api/sessions') && !url.includes(sessionId)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
sessions: [{
|
||||
id: sessionId,
|
||||
active: false,
|
||||
metadata: { name: 'Dead' }
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
if (url.endsWith(`/api/sessions/${sessionId}`)) {
|
||||
return {
|
||||
status: 200,
|
||||
data: {
|
||||
session: {
|
||||
id: sessionId,
|
||||
active: false,
|
||||
metadata: { name: 'Dead' }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new Error(`unexpected GET ${url}`)
|
||||
}
|
||||
})
|
||||
|
||||
await expect(pingPeer({
|
||||
sessionIdPrefix: 'deadbeef',
|
||||
message: 'nudge',
|
||||
accessToken: 'tok',
|
||||
apiUrl: 'http://hub.test',
|
||||
http: http as never
|
||||
})).rejects.toMatchObject({ code: 'resume_failed' })
|
||||
})
|
||||
|
||||
it('maps exit codes', () => {
|
||||
expect(exitCodeForPingPeerError(new PingPeerError('bad_args', 'x'))).toBe(2)
|
||||
expect(exitCodeForPingPeerError(new PingPeerError('resume_failed', 'x'))).toBe(3)
|
||||
expect(exitCodeForPingPeerError(new PingPeerError('timeout', 'x'))).toBe(4)
|
||||
expect(exitCodeForPingPeerError(new PingPeerError('send_failed', 'x'))).toBe(4)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,424 @@
|
||||
/**
|
||||
* Resume-if-inactive + wait-active + POST /api/sessions/:id/messages.
|
||||
*
|
||||
* Shared by `hapi ping-peer` and MCP `ping_peer`. Uses the same hub JWT flow
|
||||
* as the web app (`POST /api/auth` with CLI_API_TOKEN), scoped to the token's
|
||||
* namespace. Callers must not invent parallel auth or arbitrary hosts.
|
||||
*/
|
||||
|
||||
import axios, { type AxiosInstance } from 'axios'
|
||||
import { configuration } from '@/configuration'
|
||||
import { getAuthToken } from '@/api/auth'
|
||||
import { buildHubRequestHeaders } from '@/api/hubExtraHeaders'
|
||||
|
||||
export type PingPeerErrorCode =
|
||||
| 'bad_args'
|
||||
| 'auth_failed'
|
||||
| 'not_found'
|
||||
| 'ambiguous'
|
||||
| 'resume_failed'
|
||||
| 'timeout'
|
||||
| 'send_failed'
|
||||
|
||||
export class PingPeerError extends Error {
|
||||
readonly code: PingPeerErrorCode
|
||||
|
||||
constructor(code: PingPeerErrorCode, message: string) {
|
||||
super(message)
|
||||
this.name = 'PingPeerError'
|
||||
this.code = code
|
||||
}
|
||||
}
|
||||
|
||||
export type PingPeerSessionSummary = {
|
||||
id: string
|
||||
active: boolean
|
||||
updatedAt?: number
|
||||
metadata?: {
|
||||
name?: string
|
||||
flavor?: string | null
|
||||
piSessionId?: string
|
||||
} | null
|
||||
}
|
||||
|
||||
export type PingPeerOptions = {
|
||||
sessionIdPrefix: string
|
||||
message: string
|
||||
waitActiveSecs?: number
|
||||
apiUrl?: string
|
||||
accessToken?: string
|
||||
http?: AxiosInstance
|
||||
sleep?: (ms: number) => Promise<void>
|
||||
now?: () => number
|
||||
onProgress?: (message: string) => void
|
||||
}
|
||||
|
||||
export type PingPeerResult = {
|
||||
sessionId: string
|
||||
name: string
|
||||
resumed: boolean
|
||||
}
|
||||
|
||||
export type ListPeerSessionsOptions = {
|
||||
apiUrl?: string
|
||||
accessToken?: string
|
||||
http?: AxiosInstance
|
||||
limit?: number
|
||||
}
|
||||
|
||||
const DEFAULT_WAIT_ACTIVE_SECS = 60
|
||||
const POLL_ACTIVE_MS = 2_000
|
||||
const POLL_PI_READY_MS = 1_000
|
||||
|
||||
function defaultSleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms))
|
||||
}
|
||||
|
||||
function resolveApiUrl(apiUrl?: string): string {
|
||||
const raw = (apiUrl ?? configuration.apiUrl).trim().replace(/\/+$/, '')
|
||||
if (!raw) {
|
||||
throw new PingPeerError('bad_args', 'HAPI API URL is empty')
|
||||
}
|
||||
// Peer messaging only targets the configured hub - never accept host overrides
|
||||
// from MCP tool args (security: same hub/token/namespace only).
|
||||
return raw
|
||||
}
|
||||
|
||||
function resolveAccessToken(accessToken?: string): string {
|
||||
const token = (accessToken ?? getAuthToken()).trim()
|
||||
if (!token) {
|
||||
throw new PingPeerError('bad_args', 'CLI_API_TOKEN is required (run `hapi auth login`)')
|
||||
}
|
||||
return token
|
||||
}
|
||||
|
||||
async function exchangeJwt(
|
||||
apiUrl: string,
|
||||
accessToken: string,
|
||||
http: AxiosInstance
|
||||
): Promise<string> {
|
||||
try {
|
||||
const response = await http.post(
|
||||
`${apiUrl}/api/auth`,
|
||||
{ accessToken },
|
||||
{
|
||||
headers: buildHubRequestHeaders({ 'Content-Type': 'application/json' }),
|
||||
timeout: 10_000,
|
||||
validateStatus: () => true
|
||||
}
|
||||
)
|
||||
const token = typeof response.data?.token === 'string' ? response.data.token : ''
|
||||
if (response.status < 200 || response.status >= 300 || !token) {
|
||||
const detail = typeof response.data?.error === 'string'
|
||||
? response.data.error
|
||||
: `HTTP ${response.status}`
|
||||
throw new PingPeerError('auth_failed', `failed to exchange access token for JWT (${detail})`)
|
||||
}
|
||||
return token
|
||||
} catch (error) {
|
||||
if (error instanceof PingPeerError) {
|
||||
throw error
|
||||
}
|
||||
throw new PingPeerError(
|
||||
'auth_failed',
|
||||
`failed to exchange access token for JWT (${error instanceof Error ? error.message : String(error)})`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
function authHeaders(jwt: string): Record<string, string> {
|
||||
return buildHubRequestHeaders({
|
||||
Authorization: `Bearer ${jwt}`,
|
||||
'Content-Type': 'application/json'
|
||||
})
|
||||
}
|
||||
|
||||
export function resolveSessionByPrefix(
|
||||
sessions: PingPeerSessionSummary[],
|
||||
prefix: string
|
||||
): PingPeerSessionSummary {
|
||||
const trimmed = prefix.trim()
|
||||
if (!trimmed) {
|
||||
throw new PingPeerError('bad_args', 'session id prefix is required')
|
||||
}
|
||||
|
||||
const exact = sessions.filter((session) => session.id === trimmed)
|
||||
if (exact.length === 1) {
|
||||
return exact[0]!
|
||||
}
|
||||
|
||||
const matches = sessions.filter((session) => session.id.startsWith(trimmed))
|
||||
if (matches.length === 0) {
|
||||
throw new PingPeerError('not_found', `no session matching prefix '${trimmed}'`)
|
||||
}
|
||||
if (matches.length > 1) {
|
||||
const sample = matches.slice(0, 5).map((session) => session.id.slice(0, 8)).join(', ')
|
||||
throw new PingPeerError(
|
||||
'ambiguous',
|
||||
`prefix '${trimmed}' matches ${matches.length} sessions (${sample}${matches.length > 5 ? ', ...' : ''}); use a longer prefix`
|
||||
)
|
||||
}
|
||||
return matches[0]!
|
||||
}
|
||||
|
||||
async function listSessions(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
http: AxiosInstance,
|
||||
limit = 500
|
||||
): Promise<PingPeerSessionSummary[]> {
|
||||
const response = await http.get(
|
||||
`${apiUrl}/api/sessions`,
|
||||
{
|
||||
headers: authHeaders(jwt),
|
||||
params: { limit },
|
||||
timeout: 15_000,
|
||||
validateStatus: () => true
|
||||
}
|
||||
)
|
||||
if (response.status < 200 || response.status >= 300) {
|
||||
const detail = typeof response.data?.error === 'string'
|
||||
? response.data.error
|
||||
: `HTTP ${response.status}`
|
||||
throw new PingPeerError('auth_failed', `failed to list sessions (${detail})`)
|
||||
}
|
||||
const body = response.data
|
||||
const sessions = Array.isArray(body?.sessions)
|
||||
? body.sessions
|
||||
: Array.isArray(body)
|
||||
? body
|
||||
: null
|
||||
if (!sessions) {
|
||||
throw new PingPeerError('auth_failed', 'failed to list sessions (unexpected response)')
|
||||
}
|
||||
return sessions as PingPeerSessionSummary[]
|
||||
}
|
||||
|
||||
async function getSession(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
sessionId: string,
|
||||
http: AxiosInstance
|
||||
): Promise<PingPeerSessionSummary> {
|
||||
const response = await http.get(
|
||||
`${apiUrl}/api/sessions/${encodeURIComponent(sessionId)}`,
|
||||
{
|
||||
headers: authHeaders(jwt),
|
||||
timeout: 10_000,
|
||||
validateStatus: () => true
|
||||
}
|
||||
)
|
||||
if (response.status < 200 || response.status >= 300 || !response.data?.session) {
|
||||
const detail = typeof response.data?.error === 'string'
|
||||
? response.data.error
|
||||
: `HTTP ${response.status}`
|
||||
throw new PingPeerError('not_found', `failed to load session ${sessionId} (${detail})`)
|
||||
}
|
||||
return response.data.session as PingPeerSessionSummary
|
||||
}
|
||||
|
||||
async function resumeSession(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
sessionId: string,
|
||||
http: AxiosInstance
|
||||
): Promise<void> {
|
||||
const response = await http.post(
|
||||
`${apiUrl}/api/sessions/${encodeURIComponent(sessionId)}/resume`,
|
||||
{},
|
||||
{
|
||||
headers: authHeaders(jwt),
|
||||
timeout: 30_000,
|
||||
validateStatus: () => true
|
||||
}
|
||||
)
|
||||
if (response.data?.type === 'success') {
|
||||
return
|
||||
}
|
||||
const detail = typeof response.data?.message === 'string'
|
||||
? response.data.message
|
||||
: typeof response.data?.error === 'string'
|
||||
? response.data.error
|
||||
: typeof response.data?.code === 'string'
|
||||
? response.data.code
|
||||
: `HTTP ${response.status}`
|
||||
throw new PingPeerError('resume_failed', `resume failed: ${detail}`)
|
||||
}
|
||||
|
||||
async function waitUntilActive(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
sessionId: string,
|
||||
waitActiveSecs: number,
|
||||
http: AxiosInstance,
|
||||
sleep: (ms: number) => Promise<void>,
|
||||
now: () => number,
|
||||
onProgress?: (message: string) => void
|
||||
): Promise<void> {
|
||||
const deadline = now() + waitActiveSecs * 1000
|
||||
onProgress?.(`waiting up to ${waitActiveSecs}s for active state...`)
|
||||
while (now() < deadline) {
|
||||
const session = await getSession(apiUrl, jwt, sessionId, http)
|
||||
if (session.active) {
|
||||
return
|
||||
}
|
||||
await sleep(POLL_ACTIVE_MS)
|
||||
}
|
||||
throw new PingPeerError(
|
||||
'timeout',
|
||||
`session did not become active within ${waitActiveSecs}s; runner may have failed to spawn`
|
||||
)
|
||||
}
|
||||
|
||||
async function waitForPiReady(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
sessionId: string,
|
||||
waitActiveSecs: number,
|
||||
http: AxiosInstance,
|
||||
sleep: (ms: number) => Promise<void>,
|
||||
now: () => number,
|
||||
onProgress?: (message: string) => void
|
||||
): Promise<void> {
|
||||
// active can precede piSessionId (tiann/hapi#1143). Instant /messages before
|
||||
// get_state settles wedges (Prompt accepted / agent_start / silence).
|
||||
onProgress?.(`flavor=pi - waiting up to ${waitActiveSecs}s for metadata.piSessionId...`)
|
||||
const deadline = now() + waitActiveSecs * 1000
|
||||
while (now() < deadline) {
|
||||
const session = await getSession(apiUrl, jwt, sessionId, http)
|
||||
const piSessionId = session.metadata?.piSessionId
|
||||
if (typeof piSessionId === 'string' && piSessionId.length > 0) {
|
||||
onProgress?.(`piSessionId=${piSessionId}`)
|
||||
return
|
||||
}
|
||||
await sleep(POLL_PI_READY_MS)
|
||||
}
|
||||
throw new PingPeerError(
|
||||
'timeout',
|
||||
`piSessionId never appeared within ${waitActiveSecs}s; refusing to send (would likely wedge - see #1143)`
|
||||
)
|
||||
}
|
||||
|
||||
async function sendMessage(
|
||||
apiUrl: string,
|
||||
jwt: string,
|
||||
sessionId: string,
|
||||
message: string,
|
||||
http: AxiosInstance
|
||||
): Promise<void> {
|
||||
const response = await http.post(
|
||||
`${apiUrl}/api/sessions/${encodeURIComponent(sessionId)}/messages`,
|
||||
{ text: message },
|
||||
{
|
||||
headers: authHeaders(jwt),
|
||||
timeout: 30_000,
|
||||
validateStatus: () => true
|
||||
}
|
||||
)
|
||||
if (response.status >= 200 && response.status < 300 && response.data?.ok === true) {
|
||||
return
|
||||
}
|
||||
const detail = typeof response.data?.error === 'string'
|
||||
? response.data.error
|
||||
: typeof response.data?.code === 'string'
|
||||
? response.data.code
|
||||
: `HTTP ${response.status}`
|
||||
throw new PingPeerError('send_failed', `send failed: ${detail}`)
|
||||
}
|
||||
|
||||
export async function listPeerSessions(
|
||||
options: ListPeerSessionsOptions = {}
|
||||
): Promise<PingPeerSessionSummary[]> {
|
||||
const apiUrl = resolveApiUrl(options.apiUrl)
|
||||
const accessToken = resolveAccessToken(options.accessToken)
|
||||
const http = options.http ?? axios
|
||||
const jwt = await exchangeJwt(apiUrl, accessToken, http)
|
||||
return listSessions(apiUrl, jwt, http, options.limit ?? 200)
|
||||
}
|
||||
|
||||
export async function pingPeer(options: PingPeerOptions): Promise<PingPeerResult> {
|
||||
const prefix = options.sessionIdPrefix?.trim() ?? ''
|
||||
const message = options.message ?? ''
|
||||
if (!prefix) {
|
||||
throw new PingPeerError('bad_args', 'session id prefix is required')
|
||||
}
|
||||
if (!message) {
|
||||
throw new PingPeerError('bad_args', 'message is required')
|
||||
}
|
||||
|
||||
const waitActiveSecs = options.waitActiveSecs ?? DEFAULT_WAIT_ACTIVE_SECS
|
||||
if (!Number.isFinite(waitActiveSecs) || waitActiveSecs <= 0) {
|
||||
throw new PingPeerError('bad_args', 'waitActiveSecs must be a positive number')
|
||||
}
|
||||
|
||||
const apiUrl = resolveApiUrl(options.apiUrl)
|
||||
const accessToken = resolveAccessToken(options.accessToken)
|
||||
const http = options.http ?? axios
|
||||
const sleep = options.sleep ?? defaultSleep
|
||||
const now = options.now ?? Date.now
|
||||
const onProgress = options.onProgress
|
||||
|
||||
const jwt = await exchangeJwt(apiUrl, accessToken, http)
|
||||
const sessions = await listSessions(apiUrl, jwt, http)
|
||||
const matched = resolveSessionByPrefix(sessions, prefix)
|
||||
const name = matched.metadata?.name ?? '(unnamed)'
|
||||
onProgress?.(`resolved ${matched.id} active=${matched.active} name="${name}"`)
|
||||
|
||||
let resumed = false
|
||||
const ensureActive = async (progressMessage: string): Promise<PingPeerSessionSummary> => {
|
||||
const session = await getSession(apiUrl, jwt, matched.id, http)
|
||||
if (session.active) {
|
||||
return session
|
||||
}
|
||||
onProgress?.(progressMessage)
|
||||
await resumeSession(apiUrl, jwt, matched.id, http)
|
||||
resumed = true
|
||||
await waitUntilActive(apiUrl, jwt, matched.id, waitActiveSecs, http, sleep, now, onProgress)
|
||||
onProgress?.('session active')
|
||||
return getSession(apiUrl, jwt, matched.id, http)
|
||||
}
|
||||
|
||||
// Prefer the list snapshot for the first resume decision, then re-check before
|
||||
// send so a flip to inactive between list and POST cannot 409 (#1195).
|
||||
if (!matched.active) {
|
||||
await ensureActive('requesting resume...')
|
||||
}
|
||||
|
||||
let live = await ensureActive('session went inactive before send; requesting resume...')
|
||||
if (live.metadata?.flavor === 'pi') {
|
||||
await waitForPiReady(apiUrl, jwt, matched.id, waitActiveSecs, http, sleep, now, onProgress)
|
||||
const beforePiResume = resumed
|
||||
live = await ensureActive('session went inactive before send; requesting resume...')
|
||||
if (resumed && !beforePiResume && live.metadata?.flavor === 'pi') {
|
||||
// Fresh agent after mid-wait resume: wait for piSessionId again (#1143).
|
||||
await waitForPiReady(apiUrl, jwt, matched.id, waitActiveSecs, http, sleep, now, onProgress)
|
||||
live = await ensureActive('session went inactive before send; requesting resume...')
|
||||
}
|
||||
}
|
||||
|
||||
onProgress?.(`sending message (${message.length} chars)...`)
|
||||
await sendMessage(apiUrl, jwt, matched.id, message, http)
|
||||
|
||||
return {
|
||||
sessionId: matched.id,
|
||||
name,
|
||||
resumed
|
||||
}
|
||||
}
|
||||
|
||||
export function exitCodeForPingPeerError(error: PingPeerError): number {
|
||||
switch (error.code) {
|
||||
case 'bad_args':
|
||||
case 'auth_failed':
|
||||
case 'not_found':
|
||||
case 'ambiguous':
|
||||
return 2
|
||||
case 'resume_failed':
|
||||
return 3
|
||||
case 'timeout':
|
||||
case 'send_failed':
|
||||
return 4
|
||||
default:
|
||||
return 1
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user