diff --git a/cli/README.md b/cli/README.md index 54395fce..a08c3731 100644 --- a/cli/README.md +++ b/cli/README.md @@ -47,12 +47,21 @@ Choose a supported coding agent from your terminal and control its sessions remo - `hapi inspect-peer ` - Read-only peer metadata + recent message text (no resume). Prefer this or MCP `inspect_peer` when a user cites `[title](/sessions/)` or Copy-reference `See session "…" (/sessions/) for context`. `/sessions/` is a hub path, not a local file. Optional `--limit`. - `hapi set-mode ` - Change the permission mode of an existing session, like the web permission picker. Active sessions switch immediately; inactive ones use it on next resume. An invalid mode prints the modes allowed for that agent. -Peer messages default to `queue`: wait for the next turn, or resume an inactive -session. Use `--delivery-mode steer` to request insertion into an active Pi, -Codex, or Cursor ACP turn. MCP `ping_peer` accepts the same optional -`deliveryMode: "queue" | "steer"` field. +Peer messages default to `auto`: use the target's live state to steer an active +Codex turn when steering is available; otherwise queue, resuming an inactive +session if needed. Auto currently queues for all other agent flavors. Prefer +auto for routine collaboration when queueing is an acceptable fallback. +Use explicit `queue` for non-urgent updates that can wait. Use `steer` for +important, time-sensitive findings the peer needs to know now, not after its +current turn: a correction that invalidates its assumptions, a blocker, or a +result that changes its next action. State why it matters now and what the peer +should adjust. Steer requires insertion into an active Pi, Codex, or Cursor ACP +turn; it does not abort that turn or guarantee an immediate response. MCP `ping_peer` accepts +the same optional `deliveryMode: "auto" | "queue" | "steer"` field. Auto is +resolved by the peer sender; ordinary web/native chat delivery is unchanged. ```bash +hapi ping-peer 'The dependency is ready' --delivery-mode auto hapi ping-peer 'Run the next check' --delivery-mode queue hapi ping-peer 'Use the corrected input path' --delivery-mode steer ``` @@ -64,6 +73,9 @@ acknowledges consumption. Steer messages retain the same peer identity prefix. Steer rejects unsupported agents, idle targets, and inactive targets before sending. It reuses the queued-message Steer action: success reports `steered`, or `invoked` if the target consumed the message before the action arrived. +Auto uses Codex's `steeringActive` state and control capabilities, not just +`active` or `thinking`. If the turn changes after selection, auto uses the same +safe error handling as explicit steer; it does not resend or silently retry. If steering fails after the message was accepted, the error includes its `localId`; it may still be queued or already consumed. Inspect the target before retrying; do not automatically resend the prompt. Steering does not diff --git a/cli/src/claude/utils/startHappyServer.test.ts b/cli/src/claude/utils/startHappyServer.test.ts index 3edd9f9a..e8c27b7f 100644 --- a/cli/src/claude/utils/startHappyServer.test.ts +++ b/cli/src/claude/utils/startHappyServer.test.ts @@ -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 } : {}) } diff --git a/cli/src/claude/utils/startHappyServer.ts b/cli/src/claude/utils/startHappyServer.ts index f3e2f8ef..176fb9f6 100644 --- a/cli/src/claude/utils/startHappyServer.ts +++ b/cli/src/claude/utils/startHappyServer.ts @@ -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({ diff --git a/cli/src/claude/utils/systemPrompt.peerTools.test.ts b/cli/src/claude/utils/systemPrompt.peerTools.test.ts index a7027c6c..d2365566 100644 --- a/cli/src/claude/utils/systemPrompt.peerTools.test.ts +++ b/cli/src/claude/utils/systemPrompt.peerTools.test.ts @@ -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() } diff --git a/cli/src/codex/happyMcpStdioBridge.test.ts b/cli/src/codex/happyMcpStdioBridge.test.ts index 64d0c4e6..6b54af07 100644 --- a/cli/src/codex/happyMcpStdioBridge.test.ts +++ b/cli/src/codex/happyMcpStdioBridge.test.ts @@ -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 } }) -}) \ No newline at end of file +}) diff --git a/cli/src/codex/happyMcpStdioBridge.ts b/cli/src/codex/happyMcpStdioBridge.ts index dd295854..3788905a 100644 --- a/cli/src/codex/happyMcpStdioBridge.ts +++ b/cli/src/codex/happyMcpStdioBridge.ts @@ -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 { 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')) { diff --git a/cli/src/commands/pingPeer.test.ts b/cli/src/commands/pingPeer.test.ts index bcc150c5..a2921a75 100644 --- a/cli/src/commands/pingPeer.test.ts +++ b/cli/src/commands/pingPeer.test.ts @@ -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'] diff --git a/cli/src/commands/pingPeer.ts b/cli/src/commands/pingPeer.ts index 4b867f4b..f8f2660b 100644 --- a/cli/src/commands/pingPeer.ts +++ b/cli/src/commands/pingPeer.ts @@ -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 - hapi ping-peer --delivery-mode queue|steer + hapi ping-peer --delivery-mode auto|queue|steer hapi ping-peer --message-file hapi ping-peer --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') { diff --git a/cli/src/modules/pingPeer/deliveryMode.ts b/cli/src/modules/pingPeer/deliveryMode.ts new file mode 100644 index 00000000..c379ec56 --- /dev/null +++ b/cli/src/modules/pingPeer/deliveryMode.ts @@ -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 diff --git a/cli/src/modules/pingPeer/pingPeer.test.ts b/cli/src/modules/pingPeer/pingPeer.test.ts index a20f19ad..2f1f89c8 100644 --- a/cli/src/modules/pingPeer/pingPeer.test.ts +++ b/cli/src/modules/pingPeer/pingPeer.test.ts @@ -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 () => { diff --git a/cli/src/modules/pingPeer/pingPeer.ts b/cli/src/modules/pingPeer/pingPeer.ts index 9709820b..e6a2d813 100644 --- a/cli/src/modules/pingPeer/pingPeer.ts +++ b/cli/src/modules/pingPeer/pingPeer.ts @@ -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 { assertPeerToolsEnabled() const prefix = normalizeSessionIdPrefix(options.sessionIdPrefix ?? '') @@ -556,10 +576,11 @@ export async function pingPeer(options: PingPeerOptions): Promise { }) describe('buildSessionCitationSteerInstruction', () => { + it('guides collaboration toward auto and preserves explicit modes and retry safety', () => { + const text = buildSessionCitationSteerInstruction({ inspectTool: 'inspect_peer', pingTool: 'ping_peer' }) + expect(text).toContain(PEER_DELIVERY_MODE_DESCRIPTION) + expect(PING_PEER_TOOL_DESCRIPTION).toContain(PEER_DELIVERY_MODE_DESCRIPTION) + expect(text).toContain('auto (default)') + expect(text).toContain('active Codex turn') + expect(text).toContain('Auto currently queues for other agent flavors') + expect(text).toContain('Prefer auto for routine collaboration when queueing is an acceptable fallback') + expect(text).toContain('queue: explicitly defer non-urgent updates') + expect(text).toContain('steer: use for important, time-sensitive findings the peer needs to know now') + expect(text).toContain('not after finishing its current turn') + expect(text).toContain('State why the finding matters now and what the peer should adjust') + expect(text).toContain('Steer requires insertion') + expect(text).toContain('does not abort the current turn or guarantee an immediate response') + expect(text).toContain('do not resend automatically') + expect(text).toContain('hapi ping-peer --delivery-mode auto') + }) + it('mentions both citation forms and forbids local FS search', () => { const text = buildSessionCitationSteerInstruction({ inspectTool: 'mcp__hapi__inspect_peer', diff --git a/shared/src/sessionCitation.ts b/shared/src/sessionCitation.ts index e13e0c39..2662403f 100644 --- a/shared/src/sessionCitation.ts +++ b/shared/src/sessionCitation.ts @@ -24,12 +24,25 @@ export const INSPECT_PEER_TOOL_DESCRIPTION = 'Extract and pass it as sessionIdPrefix. /sessions/ is a hub path - do NOT Grep, Glob, or Read it as a local filesystem path. ' + 'Read-only: does not resume. Prefer this (or `hapi inspect-peer`) over JWT+curl.' +/** Shared delivery guidance for the peer MCP schemas, descriptions, and prompts. */ +export const PEER_DELIVERY_MODE_DESCRIPTION = + 'auto (default): steer an active Codex turn when steering is available; otherwise queue, resuming inactive peers. ' + + 'Auto currently queues for other agent flavors. Prefer auto for routine collaboration when queueing is an acceptable fallback. ' + + 'queue: explicitly defer non-urgent updates that can wait instead of steering. ' + + 'steer: use for important, time-sensitive findings the peer needs to know now, not after finishing its current turn ' + + '(e.g. a correction that invalidates its assumptions, a blocker, or a result that changes its next action). ' + + 'State why the finding matters now and what the peer should adjust. ' + + 'Steer requires insertion into an active Pi, Codex, or Cursor ACP turn; unsupported, idle, or inactive targets are rejected before sending. ' + + 'Steering does not abort the current turn or guarantee an immediate response. ' + + 'If send or steer is unconfirmed after submission, inspect the target before retrying; do not resend automatically.' + /** MCP `ping_peer` tool description (same citation forms as inspect_peer). */ export const PING_PEER_TOOL_DESCRIPTION = 'Send a message to another HAPI session (peer handoff / nudge). Resolves by session id prefix, resumes if inactive, then POSTs on the same hub/namespace. ' + 'When the user cites a peer via [title](/sessions/), Copy-reference prose See session "…" (/sessions/) for context, or a bare /sessions/, ' + 'extract and pass it as sessionIdPrefix. /sessions/ is a hub path - do NOT search the local filesystem for it. ' + - 'Prefer this (or `hapi ping-peer`) over reinventing JWT+curl. Targets another session - not the current chat.' + 'Prefer this (or `hapi ping-peer`) over reinventing JWT+curl. Targets another session - not the current chat. ' + + `Select deliveryMode: ${PEER_DELIVERY_MODE_DESCRIPTION}` /** Zod `.describe` for sessionIdPrefix on inspect_peer / ping_peer. */ export const SESSION_ID_PREFIX_PARAM_DESCRIPTION = @@ -158,7 +171,8 @@ export function buildSessionCitationSteerInstruction(tools: SessionCitationSteer `/sessions/ is a HAPI hub path, not a local filesystem path - do not Grep, Glob, or Read it as a file. ` + `Call "${tools.inspectTool}" with sessionIdPrefix= to read metadata and recent messages; ` + `call "${tools.pingTool}" with sessionIdPrefix= and a message to nudge or hand off. ` + - `Prefer these over JWT+curl. Shell fallbacks: hapi inspect-peer / hapi ping-peer .` + `Select deliveryMode: ${PEER_DELIVERY_MODE_DESCRIPTION} ` + + `Prefer these over JWT+curl. Shell fallbacks: hapi inspect-peer / hapi ping-peer --delivery-mode auto.` if (tools.listPeersTool) { text += ` To discover peers without a citation, call "${tools.listPeersTool}" ` +