wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-11-08 01:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cli): add Codex-aware auto peer delivery

Browse Source 
Default peer CLI and MCP messages to auto, steering eligible Codex turns and queueing otherwise. Preserve explicit delivery modes and uncertain-send safety, and clarify steer guidance for important time-sensitive findings.
This commit is contained in:
weishu
2026-10-29 22:02:43 +08:00
parent f48a6696cf
commit 9384ce031b
13 changed files with 270 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/claude/utils/startHappyServer.test.ts
+6 -1
View File
@@ -131,6 +131,10 @@ describe('startHappyServer skill_lookup', () => {
it.each([
[undefined, 'accepted'],
[undefined, 'steered'],
['auto', 'accepted'],
['auto', 'steered'],
['auto', 'invoked'],
['queue', 'accepted'],
['steer', 'steered'],
['steer', 'invoked']
@@ -143,7 +147,8 @@ describe('startHappyServer skill_lookup', () => {
const mcp = await connect(false)
const tools = await mcp.listTools()
const tool = tools.tools.find((entry) => entry.name === 'ping_peer')
expect(tool?.inputSchema.properties?.deliveryMode).toMatchObject({ enum: ['queue', 'steer'] })
expect(tool?.inputSchema.properties?.deliveryMode).toMatchObject({ enum: ['auto', 'queue', 'steer'] })
expect(tool?.description).toContain('auto (default)')
const result = await mcp.callTool({
name: 'ping_peer',
arguments: { sessionIdPrefix: 'peer-id', message: 'hello', ...(deliveryMode ? { deliveryMode } : {}) }
cli/src/claude/utils/startHappyServer.ts
+4 -5
View File
@@ -8,7 +8,7 @@ import { createServer, type IncomingMessage } from "node:http";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { AddressInfo } from "node:net";
import { z } from "zod";
import { MessageDeliveryModeSchema, type MessageDeliveryMode } from '@hapi/protocol';
import { PeerDeliveryModeSchema, type PeerDeliveryMode } from '@/modules/pingPeer/deliveryMode';
import { logger } from "@/ui/logger";
import { ApiSessionClient } from "@/api/apiSession";
import { randomUUID } from "node:crypto";
@@ -25,6 +25,7 @@ import { resolveSkill } from "@/modules/common/skills";
import { isPeerToolsDisabled } from "@/modules/common/peerTools";
import {
INSPECT_PEER_TOOL_DESCRIPTION,
PEER_DELIVERY_MODE_DESCRIPTION,
PING_PEER_TOOL_DESCRIPTION,
SESSION_ID_PREFIX_PARAM_DESCRIPTION,
} from '@hapi/protocol/sessionCitation'
@@ -134,9 +135,7 @@ function createHapiMcpServer(
const pingPeerInputSchema: z.ZodTypeAny = z.object({
sessionIdPrefix: z.string().trim().min(1).describe(SESSION_ID_PREFIX_PARAM_DESCRIPTION),
message: z.string().min(1).describe('Message text to deliver to the target session'),
deliveryMode: MessageDeliveryModeSchema.optional().describe(
'queue (default): next turn, resuming if needed. steer: insert into an active Pi, Codex, or Cursor ACP turn; idle/inactive targets are rejected. If steer fails after send, inspect the target before retrying; do not resend automatically.'
),
deliveryMode: PeerDeliveryModeSchema.optional().describe(PEER_DELIVERY_MODE_DESCRIPTION),
});
const maxInlineMediaBytes = 25 * 1024 * 1024;
@@ -327,7 +326,7 @@ function createHapiMcpServer(
description: PING_PEER_TOOL_DESCRIPTION,
title: 'Ping Peer Session',
inputSchema: pingPeerInputSchema,
}, async (args: { sessionIdPrefix: string; message: string; deliveryMode?: MessageDeliveryMode }) => {
}, async (args: { sessionIdPrefix: string; message: string; deliveryMode?: PeerDeliveryMode }) => {
logger.debug('[hapiMCP] ping_peer:', args.sessionIdPrefix);
try {
const result = await pingPeer({
cli/src/claude/utils/systemPrompt.peerTools.test.ts
+10
View File
@@ -42,6 +42,9 @@ describe('peer-tools citation steer across flavors', () => {
expect(opencode.OPENCODE_NATIVE_TOOL_INSTRUCTION).toContain('are disabled on this machine')
expect(opencode.OPENCODE_NATIVE_TOOL_INSTRUCTION).not.toContain('Call "hapi_inspect_peer"')
for (const prompt of [claudePrompt, codex.TITLE_INSTRUCTION, opencode.TITLE_INSTRUCTION, opencode.OPENCODE_NATIVE_TOOL_INSTRUCTION]) {
expect(prompt).not.toContain('auto (default)')
}
} finally {
restore()
}
@@ -54,6 +57,13 @@ describe('peer-tools citation steer across flavors', () => {
expect(codex.TITLE_INSTRUCTION).toContain('Call "functions.hapi__inspect_peer"')
expect(opencode.TITLE_INSTRUCTION).toContain('Call "hapi_inspect_peer"')
expect(opencode.OPENCODE_NATIVE_TOOL_INSTRUCTION).toContain('Call "hapi_inspect_peer"')
for (const prompt of [claude.getSystemPrompt(), codex.TITLE_INSTRUCTION, opencode.TITLE_INSTRUCTION, opencode.OPENCODE_NATIVE_TOOL_INSTRUCTION]) {
expect(prompt).toContain('auto (default)')
expect(prompt).toContain('Prefer auto for routine collaboration')
expect(prompt).toContain('steer: use for important, time-sensitive findings the peer needs to know now')
expect(prompt).toContain('does not abort the current turn or guarantee an immediate response')
expect(prompt).toContain('do not resend automatically')
}
} finally {
restore()
}
cli/src/codex/happyMcpStdioBridge.test.ts
+3 -2
View File
@@ -140,12 +140,13 @@ describe('runHappyMcpStdioBridge tool forwarding', () => {
])
})
it.each(['queue', 'steer'])('advertises and forwards ping_peer delivery mode %s', async (deliveryMode) => {
it.each(['auto', 'queue', 'steer'])('advertises and forwards ping_peer delivery mode %s', async (deliveryMode) => {
await runHappyMcpStdioBridge(['--url', 'http://127.0.0.1:43006', '--tools', 'ping_peer'])
const schema = harness.configs.get('ping_peer')?.inputSchema as {
parse(input: unknown): unknown
}
const args = { sessionIdPrefix: 'peer-id', message: 'hello', deliveryMode }
expect(harness.configs.get('ping_peer')?.description).toContain('auto (default)')
expect(schema.parse(args)).toEqual(args)
expect(() => schema.parse({ ...args, deliveryMode: 'interrupt' })).toThrow()
await harness.tools.get('ping_peer')?.(args)
@@ -238,4 +239,4 @@ describe('runHappyMcpStdioBridge tool forwarding', () => {
delete process.env.HAPI_DISABLE_PEER_TOOLS
}
})
})
})
cli/src/codex/happyMcpStdioBridge.ts
+3 -4
View File
@@ -16,10 +16,11 @@ import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
import { z } from 'zod';
import { MessageDeliveryModeSchema } from '@hapi/protocol';
import { PeerDeliveryModeSchema } from '@/modules/pingPeer/deliveryMode';
import { DISPLAY_IMAGE_PROMPT_CURSOR, DISPLAY_MEDIA_PROMPT_CURSOR, DISPLAY_VIDEO_PROMPT_CURSOR } from '@/modules/common/displayImagePrompt';
import {
INSPECT_PEER_TOOL_DESCRIPTION,
PEER_DELIVERY_MODE_DESCRIPTION,
PING_PEER_TOOL_DESCRIPTION,
SESSION_ID_PREFIX_PARAM_DESCRIPTION,
} from '@hapi/protocol/sessionCitation';
@@ -213,9 +214,7 @@ export async function runHappyMcpStdioBridge(argv: string[]): Promise<void> {
const pingPeerInputSchema: z.ZodTypeAny = z.object({
sessionIdPrefix: z.string().trim().min(1).describe(SESSION_ID_PREFIX_PARAM_DESCRIPTION),
message: z.string().min(1).describe('Message text to deliver to the target session'),
deliveryMode: MessageDeliveryModeSchema.optional().describe(
'queue (default): next turn, resuming if needed. steer: insert into an active Pi, Codex, or Cursor ACP turn; idle/inactive targets are rejected. If steer fails after send, inspect the target before retrying; do not resend automatically.'
),
deliveryMode: PeerDeliveryModeSchema.optional().describe(PEER_DELIVERY_MODE_DESCRIPTION),
});
if (toolNames.has('ping_peer')) {
cli/src/commands/pingPeer.test.ts
+5 -1
View File
@@ -33,7 +33,7 @@ describe('parsePingPeerArgs', () => {
expect(() => parsePingPeerArgs(['--host', 'evil'])).toThrow(PingPeerError)
})
it.each(['queue', 'steer'] as const)('parses delivery mode %s in both flag forms', (mode) => {
it.each(['auto', 'queue', 'steer'] as const)('parses delivery mode %s in both flag forms', (mode) => {
for (const flags of [['--delivery-mode', mode], [`--delivery-mode=${mode}`]]) {
expect(parsePingPeerArgs(['abc', 'hello', ...flags]).deliveryMode).toBe(mode)
}
@@ -57,6 +57,10 @@ describe('handlePingPeerCommand receipts', () => {
it.each([
[undefined, 'accepted'],
[undefined, 'steered'],
['auto', 'accepted'],
['auto', 'steered'],
['auto', 'invoked'],
['queue', 'accepted'],
['steer', 'steered'],
['steer', 'invoked']
cli/src/commands/pingPeer.ts
+14 -7
View File
@@ -1,6 +1,6 @@
import { readFile } from 'node:fs/promises'
import chalk from 'chalk'
import type { MessageDeliveryMode } from '@hapi/protocol'
import { PeerDeliveryModeSchema, type PeerDeliveryMode } from '@/modules/pingPeer/deliveryMode'
import { initializeToken } from '@/ui/tokenInit'
import {
PingPeerError,
@@ -19,7 +19,7 @@ type ParsedPingPeerArgs = {
message?: string
messageFile?: string
waitActiveSecs?: number
deliveryMode?: MessageDeliveryMode
deliveryMode?: PeerDeliveryMode
}
function showHelp(): void {
@@ -28,7 +28,7 @@ ${chalk.bold('hapi ping-peer')} - Resume a HAPI session (if needed) and send it
${chalk.bold('Usage:')}
hapi ping-peer <session-id-prefix> <message-text>
hapi ping-peer <session-id-prefix> <message-text> --delivery-mode queue|steer
hapi ping-peer <session-id-prefix> <message-text> --delivery-mode auto|queue|steer
hapi ping-peer <session-id-prefix> --message-file <path>
hapi ping-peer <session-id-prefix> --message-file - # read message from stdin
hapi ping-peer --list
@@ -37,7 +37,13 @@ ${chalk.bold('Notes:')}
Do not reinvent JWT + curl for peer handoffs. Prefer this command or MCP ping_peer / list_peers.
Resolves by id prefix (8 chars OK). Same hub token/namespace as this CLI.
Inactive sessions are resumed via POST /api/sessions/:id/resume, then messaged.
Delivery defaults to queue (next turn). Steer requires an active turn in Pi, Codex,
Delivery defaults to auto: steer a steerable active Codex turn, otherwise queue.
Auto resumes inactive targets; other agent flavors currently queue in auto mode.
Use auto for routine collaboration when queueing is an acceptable fallback.
Use queue for non-urgent updates that can wait. Use steer for important findings
the peer needs to know now to adjust its current work, not after its current turn.
State why the finding matters now and what the peer should adjust.
Explicit steer requires an active turn in Pi, Codex,
or Cursor ACP; it neither resumes inactive targets nor sends to idle targets.
Steer uses the existing queued-message action. If it fails after send, inspect the
target before retrying: the accepted message may still be queued or consumed.
@@ -70,13 +76,14 @@ export function parsePingPeerArgs(args: string[]): ParsedPingPeerArgs {
}
if (arg === '--delivery-mode' || arg.startsWith('--delivery-mode=')) {
const value = arg === '--delivery-mode' ? args[++i] : arg.slice('--delivery-mode='.length)
if (value !== 'queue' && value !== 'steer') {
throw new PingPeerError('bad_args', '--delivery-mode must be queue or steer')
const parsed = PeerDeliveryModeSchema.safeParse(value)
if (!parsed.success) {
throw new PingPeerError('bad_args', '--delivery-mode must be auto, queue or steer')
}
if (result.deliveryMode !== undefined) {
throw new PingPeerError('bad_args', '--delivery-mode may only be specified once')
}
result.deliveryMode = value
result.deliveryMode = parsed.data
continue
}
if (arg === '--message-file') {
cli/src/modules/pingPeer/deliveryMode.ts
+7
View File
@@ -0,0 +1,7 @@
import { z } from 'zod'
import { MessageDeliveryModeSchema } from '@hapi/protocol'
// Auto is a peer-sender policy, not a hub/wire delivery mode. Resolve it before
// submission so ordinary chat sends and persisted queue semantics stay unchanged.
export const PeerDeliveryModeSchema = z.enum(['auto', ...MessageDeliveryModeSchema.options])
export type PeerDeliveryMode = z.infer<typeof PeerDeliveryModeSchema>
cli/src/modules/pingPeer/pingPeer.test.ts
+131 -41
View File
@@ -1,4 +1,6 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { MessageDeliveryModeSchema } from '@hapi/protocol'
import { PeerDeliveryModeSchema } from './deliveryMode'
import {
PingPeerError,
exitCodeForPingPeerError,
@@ -137,7 +139,7 @@ describe('pingPeer', () => {
expect(http.post).toHaveBeenCalledTimes(2)
})
it('resumes an inactive session, waits for active, then sends', async () => {
it.each([undefined, 'auto', 'queue'] as const)('resumes an inactive session with %s delivery, waits for active, then sends', async (deliveryMode) => {
const sessionId = 'aaaaaaaa-1111-1111-1111-111111111111'
let active = false
let polls = 0
@@ -192,6 +194,7 @@ describe('pingPeer', () => {
const result = await pingPeer({
sessionIdPrefix: 'aaaaaaaa',
message: 'wake up',
deliveryMode,
accessToken: 'tok',
apiUrl: 'http://hub.test',
waitActiveSecs: 10,
@@ -414,10 +417,10 @@ describe('pingPeer delivery modes', () => {
function setup(session = active, steerResult: MockResponse | Error = {
status: 200, data: { status: 'steered' }
}) {
}, listedSession = session) {
const http = createHttpMock({
get: (url) => url.endsWith('/api/sessions')
? { status: 200, data: { sessions: [session] } }
? { status: 200, data: { sessions: [listedSession] } }
: { status: 200, data: { session } },
post: (url) => {
if (url.endsWith('/api/auth')) return { status: 200, data: { token: 'jwt' } }
@@ -438,9 +441,14 @@ describe('pingPeer delivery modes', () => {
}
}
it.each([undefined, 'queue'] as const)('keeps %s delivery on the queue path', async (deliveryMode) => {
it('keeps auto out of the hub wire delivery modes', () => {
expect(PeerDeliveryModeSchema.parse('auto')).toBe('auto')
expect(MessageDeliveryModeSchema.safeParse('auto').success).toBe(false)
})
it('honors explicit queue even when a Codex turn is steerable', async () => {
const { http, options } = setup()
const result = await pingPeer({ ...options, deliveryMode })
const result = await pingPeer({ ...options, deliveryMode: 'queue' })
expect(result.delivery).toBe('accepted')
expect(result.localId).toMatch(/^ping-peer-[0-9a-f-]{36}$/)
expect(http.post.mock.calls.map(([url]) => url)).toEqual([
@@ -449,6 +457,85 @@ describe('pingPeer delivery modes', () => {
expect(http.post.mock.calls[1]?.[1]).toEqual({ text: options.message, localId: result.localId })
})
it.each([undefined, 'auto'] as const)('steers an active Codex peer with %s delivery', async (deliveryMode) => {
const { http, options } = setup()
const onProgress = vi.fn()
const result = await pingPeer({ ...options, deliveryMode, onProgress })
expect(result.delivery).toBe('steered')
expect(http.post.mock.calls.map(([url]) => url)).toEqual([
'http://hub.test/api/auth',
`http://hub.test/api/sessions/${sessionId}/messages`,
`http://hub.test/api/sessions/${sessionId}/messages/${result.localId}/steer`
])
expect(http.post.mock.calls[1]?.[1]).toEqual({ text: options.message, localId: result.localId })
expect(onProgress).toHaveBeenCalledWith(expect.stringContaining('steer via auto'))
})
it.each([
{ name: 'idle Codex', session: { ...active, thinking: false, agentState: { steeringActive: false } } },
{ name: 'thinking Codex without a steerable turn', session: { ...active, agentState: { steeringActive: false } } },
{ name: 'Codex without steering state', session: { ...active, agentState: null } },
{ name: 'local Codex without concurrent control', session: { ...active, agentState: { steeringActive: true, controlledByUser: true } } },
{ name: 'unknown flavor', session: { ...active, metadata: null } },
{ name: 'Pi', session: { ...active, metadata: { flavor: 'pi', piSessionId: 'pi-thread' } } },
{ name: 'Cursor ACP', session: { ...active, metadata: { flavor: 'cursor', cursorSessionProtocol: 'acp' as const } } },
{ name: 'legacy Cursor', session: { ...active, metadata: { flavor: 'cursor', cursorSessionProtocol: 'stream-json' as const } } },
{ name: 'Claude', session: { ...active, metadata: { flavor: 'claude' } } }
])('auto queues for $name', async ({ session }) => {
const { http, options } = setup(session)
expect((await pingPeer({ ...options, deliveryMode: 'auto' })).delivery).toBe('accepted')
expect(http.post.mock.calls.map(([url]) => url)).toEqual([
'http://hub.test/api/auth', `http://hub.test/api/sessions/${sessionId}/messages`
])
})
it('auto steers concurrent Codex when thinking is false but steeringActive is true', async () => {
const { options } = setup({
...active, thinking: false,
agentState: { steeringActive: true, controlledByUser: true },
metadata: { flavor: 'codex', capabilities: { concurrentClients: true } }
})
expect((await pingPeer({ ...options, deliveryMode: 'auto' })).delivery).toBe('steered')
})
it.each([true, false])('auto uses live steeringActive=%s rather than the list snapshot', async (steeringActive) => {
const { http, options } = setup(
{ ...active, agentState: { steeringActive } },
undefined,
{ ...active, agentState: { steeringActive: !steeringActive } }
)
expect((await pingPeer({ ...options, deliveryMode: 'auto' })).delivery).toBe(steeringActive ? 'steered' : 'accepted')
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/steer'))).toHaveLength(steeringActive ? 1 : 0)
})
it.each([true, false])('auto re-evaluates Codex steeringActive=%s after resume', async (steeringActive) => {
let resumed = false
const http = createHttpMock({
get: (url) => url.endsWith('/api/sessions')
? { status: 200, data: { sessions: [{ ...active, active: false }] } }
: { status: 200, data: { session: { ...active, active: resumed, agentState: { steeringActive: resumed && steeringActive } } } },
post: (url) => {
if (url.endsWith('/api/auth')) return { status: 200, data: { token: 'jwt' } }
if (url.endsWith('/resume')) {
resumed = true
return { status: 200, data: { type: 'success', sessionId } }
}
if (url.endsWith('/messages')) return { status: 200, data: { ok: true } }
if (url.endsWith('/steer')) return { status: 200, data: { status: 'steered' } }
throw new Error(`Unexpected POST ${url}`)
}
})
const result = await pingPeer({
sessionIdPrefix: sessionId, message: 'update', deliveryMode: 'auto',
apiUrl: 'http://hub.test', accessToken: 'tok', http: http as never
})
expect(result.resumed).toBe(true)
expect(result.delivery).toBe(steeringActive ? 'steered' : 'accepted')
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/resume'))).toHaveLength(1)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/messages'))).toHaveLength(1)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/steer'))).toHaveLength(steeringActive ? 1 : 0)
})
it.each([
{ flavor: 'codex' },
{ flavor: 'pi', piSessionId: 'pi-thread' },
@@ -467,9 +554,9 @@ describe('pingPeer delivery modes', () => {
expect(http.post).toHaveBeenCalledTimes(3)
})
it('reports a message consumed before the steer without claiming it was steered', async () => {
it.each([undefined, 'auto', 'steer'] as const)('reports a message consumed before %s steer without claiming it was steered', async (deliveryMode) => {
const { options } = setup(active, { status: 200, data: { status: 'invoked' } })
expect((await pingPeer({ ...options, deliveryMode: 'steer' })).delivery).toBe('invoked')
expect((await pingPeer({ ...options, deliveryMode })).delivery).toBe('invoked')
})
it.each([
@@ -522,49 +609,52 @@ describe('pingPeer delivery modes', () => {
expect(http.post.mock.calls.map(([url]) => url)).toEqual(['http://hub.test/api/auth'])
})
it.each([
{ status: 200, data: { status: 'failed', error: 'Message not in queue' } },
{ status: 409, data: { error: 'Session inactive' } },
new Error('Request timed out')
])('reports accepted but unconfirmed steer without resending: %j', async (failure) => {
const { http, options } = setup(active, failure)
await expect(pingPeer({ ...options, deliveryMode: 'steer' })).rejects.toThrow(
/Message accepted \(localId=ping-peer-.+\), but steer was not confirmed: .*Do not resend automatically/
)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/messages'))).toHaveLength(1)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/steer'))).toHaveLength(1)
})
it('validates delivery mode before making requests', async () => {
const { http, options } = setup()
await expect(pingPeer({ ...options, deliveryMode: 'interrupt' as never })).rejects.toThrow(
'deliveryMode must be queue or steer'
'deliveryMode must be auto, queue or steer'
)
expect(http.post).not.toHaveBeenCalled()
expect(http.get).not.toHaveBeenCalled()
})
it('does not steer or resend when the initial send times out', async () => {
const { http, options } = setup()
http.post.mockImplementationOnce(async () => ({ status: 200, data: { token: 'jwt' } }))
http.post.mockImplementationOnce(async () => { throw new Error('Request timed out') })
await expect(pingPeer({ ...options, deliveryMode: 'steer' })).rejects.toThrow(
/Send not confirmed \(localId=ping-peer-.+\).*Do not resend automatically/
)
expect(http.post).toHaveBeenCalledTimes(2)
})
describe.each([undefined, 'auto', 'steer'] as const)('selected steer with %s delivery', (deliveryMode) => {
it.each([
{ status: 200, data: { status: 'failed', error: 'No active turn' } },
{ status: 200, data: { status: 'failed', error: 'Message not in queue' } },
{ status: 409, data: { error: 'Session inactive' } },
new Error('Request timed out')
])('reports accepted but unconfirmed steer without fallback or resending: %j', async (failure) => {
const { http, options } = setup(active, failure)
await expect(pingPeer({ ...options, deliveryMode })).rejects.toThrow(
/Message accepted \(localId=ping-peer-.+\), but steer was not confirmed: .*Do not resend automatically/
)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/messages'))).toHaveLength(1)
expect(http.post.mock.calls.filter(([url]) => url.endsWith('/steer'))).toHaveLength(1)
})
it.each([
{ status: 500, data: { error: 'Publish failed' } },
{ status: 200, data: { ok: false } }
])('retains the localId when the initial send response does not confirm acceptance: %j', async (response) => {
const { http, options } = setup()
http.post.mockImplementationOnce(async () => ({ status: 200, data: { token: 'jwt' } }))
http.post.mockImplementationOnce(async () => response)
await expect(pingPeer({ ...options, deliveryMode: 'steer' })).rejects.toThrow(
/Send not confirmed \(localId=ping-peer-.+\).*Do not resend automatically/
)
expect(http.post).toHaveBeenCalledTimes(2)
it('does not steer or resend when the initial send times out', async () => {
const { http, options } = setup()
http.post.mockImplementationOnce(async () => ({ status: 200, data: { token: 'jwt' } }))
http.post.mockImplementationOnce(async () => { throw new Error('Request timed out') })
await expect(pingPeer({ ...options, deliveryMode })).rejects.toThrow(
/Send not confirmed \(localId=ping-peer-.+\).*Do not resend automatically/
)
expect(http.post).toHaveBeenCalledTimes(2)
})
it.each([
{ status: 500, data: { error: 'Publish failed' } },
{ status: 200, data: { ok: false } }
])('retains the localId when the initial send response does not confirm acceptance: %j', async (response) => {
const { http, options } = setup()
http.post.mockImplementationOnce(async () => ({ status: 200, data: { token: 'jwt' } }))
http.post.mockImplementationOnce(async () => response)
await expect(pingPeer({ ...options, deliveryMode })).rejects.toThrow(
/Send not confirmed \(localId=ping-peer-.+\).*Do not resend automatically/
)
expect(http.post).toHaveBeenCalledTimes(2)
})
})
it('does not attempt steering when a queue submission fails', async () => {
cli/src/modules/pingPeer/pingPeer.ts
+35 -17
View File
@@ -23,6 +23,7 @@ import { configuration } from '@/configuration'
import { isPeerToolsDisabled } from '@/modules/common/peerTools'
import { getAuthToken } from '@/api/auth'
import { buildHubRequestHeaders } from '@/api/hubExtraHeaders'
import { PeerDeliveryModeSchema, type PeerDeliveryMode } from './deliveryMode'
export type PingPeerErrorCode =
| 'bad_args'
@@ -81,7 +82,7 @@ export type PingPeerSessionSummary = {
export type PingPeerOptions = {
sessionIdPrefix: string
message: string
deliveryMode?: MessageDeliveryMode
deliveryMode?: PeerDeliveryMode
waitActiveSecs?: number
apiUrl?: string
accessToken?: string
@@ -546,6 +547,25 @@ export function formatPeerSessionsList(
return rows.join('\n')
}
function getSteeringUnavailableReason(session: PingPeerSessionSummary): string | null {
if (!session.active) {
return 'Steering requires an active turn; target is inactive, no message sent. Use queue to resume it'
}
if (!isSteeringSupportedForSession(session.metadata)) {
return 'Steering is only supported for Pi, Codex, and Cursor ACP sessions; no message sent'
}
if (session.agentState?.controlledByUser && !session.metadata?.capabilities?.concurrentClients) {
return 'Steering is only available for remote sessions; no message sent'
}
const steeringActive = session.metadata?.flavor === 'pi'
? session.thinking
: session.agentState?.steeringActive === true
if (!steeringActive) {
return 'Steering requires an active turn; target is idle, no message sent. Use queue to start a turn'
}
return null
}
export async function pingPeer(options: PingPeerOptions): Promise<PingPeerResult> {
assertPeerToolsEnabled()
const prefix = normalizeSessionIdPrefix(options.sessionIdPrefix ?? '')
@@ -556,10 +576,11 @@ export async function pingPeer(options: PingPeerOptions): Promise<PingPeerResult
if (!message) {
throw new PingPeerError('bad_args', 'message is required')
}
const deliveryMode = options.deliveryMode ?? 'queue'
if (deliveryMode !== 'queue' && deliveryMode !== 'steer') {
throw new PingPeerError('bad_args', 'deliveryMode must be queue or steer')
const parsedDeliveryMode = PeerDeliveryModeSchema.safeParse(options.deliveryMode ?? 'auto')
if (!parsedDeliveryMode.success) {
throw new PingPeerError('bad_args', 'deliveryMode must be auto, queue or steer')
}
const requestedDeliveryMode = parsedDeliveryMode.data
const waitActiveSecs = options.waitActiveSecs ?? DEFAULT_WAIT_ACTIVE_SECS
if (!Number.isFinite(waitActiveSecs) || waitActiveSecs <= 0) {
@@ -585,7 +606,7 @@ export async function pingPeer(options: PingPeerOptions): Promise<PingPeerResult
if (session.active) {
return session
}
if (deliveryMode === 'steer') {
if (requestedDeliveryMode === 'steer') {
throw new PingPeerError('bad_args', 'Steering requires an active turn; target is inactive, no message sent. Use queue to resume it')
}
onProgress?.(progressMessage)
@@ -614,20 +635,17 @@ export async function pingPeer(options: PingPeerOptions): Promise<PingPeerResult
}
}
if (deliveryMode === 'steer' && !isSteeringSupportedForSession(live.metadata)) {
throw new PingPeerError('bad_args', 'Steering is only supported for Pi, Codex, and Cursor ACP sessions; no message sent')
}
if (deliveryMode === 'steer' && live.agentState?.controlledByUser && !live.metadata?.capabilities?.concurrentClients) {
throw new PingPeerError('bad_args', 'Steering is only available for remote sessions; no message sent')
}
const steeringActive = live.metadata?.flavor === 'pi'
? live.thinking
: live.agentState?.steeringActive === true
if (deliveryMode === 'steer' && !steeringActive) {
throw new PingPeerError('bad_args', 'Steering requires an active turn; target is idle, no message sent. Use queue to start a turn')
// Use the live snapshot, not the discovery list. Auto initially steers only
// Codex targets; other flavors retain queue unless steer is explicit.
const steeringError = getSteeringUnavailableReason(live)
const deliveryMode: MessageDeliveryMode = requestedDeliveryMode === 'auto'
? (live.metadata?.flavor === 'codex' && steeringError === null ? 'steer' : 'queue')
: requestedDeliveryMode
if (deliveryMode === 'steer' && steeringError !== null) {
throw new PingPeerError('bad_args', steeringError)
}
onProgress?.(`sending message (${message.length} chars, ${deliveryMode})...`)
onProgress?.(`sending message (${message.length} chars, ${deliveryMode}${requestedDeliveryMode === 'auto' ? ' via auto' : ''})...`)
const localId = createPeerMessageLocalId()
const delivery = await sendMessage(apiUrl, jwt, matched.id, message, localId, http, deliveryMode)