diff --git a/pi/src/config.ts b/pi/src/config.ts index 2139e934a..6dc8b3cba 100644 --- a/pi/src/config.ts +++ b/pi/src/config.ts @@ -30,9 +30,12 @@ export type Config = z.infer; export const RuntimeConfigSchema = ConfigSchema.omit({ engine_url: true }); export type RuntimeConfig = z.infer; -/** JSON Schema published to the configuration worker. */ +/** JSON Schema published to the configuration worker. The registry validator + * has no `$schema` meta-schema, so strip the draft-2020-12 `$schema` key. */ export function runtimeJsonSchema(): Record { - return z.toJSONSchema(RuntimeConfigSchema) as Record; + const out = z.toJSONSchema(RuntimeConfigSchema) as Record; + delete out.$schema; + return out; } /** The runtime slice of a full config, for use as `initial_value`. */ diff --git a/pi/src/configuration.ts b/pi/src/configuration.ts index d40496535..23866daac 100644 --- a/pi/src/configuration.ts +++ b/pi/src/configuration.ts @@ -62,10 +62,18 @@ export async function fetchRuntime(iii: ISdk): Promise { */ export async function bindConfigTrigger(iii: ISdk, onChange: () => Promise): Promise { await onChange(); - iii.registerFunction(CONFIG_FN_ID, async () => { - await onChange(); - return null; - }); + iii.registerFunction( + CONFIG_FN_ID, + async () => { + await onChange(); + return null; + }, + { + description: 'Internal: reload pi configuration when it changes.', + request_format: { type: 'object', properties: {} }, + response_format: { type: 'null' }, + }, + ); iii.registerTrigger({ type: 'configuration', function_id: CONFIG_FN_ID, diff --git a/pi/src/run.ts b/pi/src/run.ts index d5c9c2085..ff9d1a6a7 100644 --- a/pi/src/run.ts +++ b/pi/src/run.ts @@ -84,9 +84,87 @@ const SteerPayloadSchema = z.object({ prompt: z.string().describe('Instruction to inject into the live run'), }); -const RUN_REQUEST_FORMAT = z.toJSONSchema(RunPayloadSchema); -const SESSION_ID_FORMAT = z.toJSONSchema(SessionIdSchema); -const STEER_REQUEST_FORMAT = z.toJSONSchema(SteerPayloadSchema); +// The registry's publish validator has no `$schema` meta-schema registered, so +// the draft-2020-12 `$schema` key z.toJSONSchema stamps at the root fails +// validation. Strip it; the schema body is what the engine + registry consume. +function jsonSchema(schema: z.ZodType): Record { + const out = z.toJSONSchema(schema) as Record; + delete out.$schema; + return out; +} + +const RUN_REQUEST_FORMAT = jsonSchema(RunPayloadSchema); +const SESSION_ID_FORMAT = jsonSchema(SessionIdSchema); +const STEER_REQUEST_FORMAT = jsonSchema(SteerPayloadSchema); + +const UsageSchema = z.object({ + input_tokens: z.number(), + output_tokens: z.number(), + cache_read_tokens: z.number().optional(), + cache_write_tokens: z.number().optional(), +}); + +const RunResultSchema = z.object({ + session_id: z.string(), + pi_session_id: z.string().nullable().optional(), + result: z.string().optional(), + stop_reason: z.string().optional(), + is_error: z.boolean().optional(), + num_turns: z.number().optional(), + total_cost_usd: z.number().optional(), + usage: UsageSchema.nullable().optional(), + busy: z.boolean().optional(), + reason: z.string().optional(), +}); +const StartResultSchema = z.object({ + session_id: z.string(), + started: z.boolean(), + busy: z.boolean().optional(), + reason: z.string().optional(), +}); +const SteerResultSchema = z.object({ + session_id: z.string(), + steered: z.boolean(), + reason: z.string().optional(), +}); +const FollowUpResultSchema = z.object({ + session_id: z.string(), + queued: z.boolean(), + reason: z.string().optional(), +}); +const StopResultSchema = z.object({ + session_id: z.string(), + stopped: z.boolean(), + reason: z.string().optional(), +}); +const SessionRecordSchema = z.object({ + session_id: z.string(), + pi_session_id: z.string().nullable(), + session_file: z.string().nullable(), + cwd: z.string(), + model: z.string(), + status: z.enum(['working', 'done', 'error']), + turns: z.number(), + total_cost_usd: z.number(), + usage: UsageSchema.nullable(), + updated_at_ms: z.number(), +}); +const StatusResultSchema = z.object({ + session_id: z.string(), + live: z.boolean(), + record: SessionRecordSchema.nullable(), +}); +const SessionsResultSchema = z.object({ + sessions: z.array(SessionRecordSchema), +}); + +const RUN_RESPONSE_FORMAT = jsonSchema(RunResultSchema); +const START_RESPONSE_FORMAT = jsonSchema(StartResultSchema); +const STEER_RESPONSE_FORMAT = jsonSchema(SteerResultSchema); +const FOLLOWUP_RESPONSE_FORMAT = jsonSchema(FollowUpResultSchema); +const STOP_RESPONSE_FORMAT = jsonSchema(StopResultSchema); +const STATUS_RESPONSE_FORMAT = jsonSchema(StatusResultSchema); +const SESSIONS_RESPONSE_FORMAT = jsonSchema(SessionsResultSchema); type LiveRun = { session: AgentSession }; const live = new Map(); @@ -318,6 +396,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E description: 'Run one Pi coding-agent turn and wait for the result. Accepts `prompt` or a `messages` array; streams raw Pi events onto pi::events, AgentEvent frames onto agent::events, and returns {session_id, result, usage, total_cost_usd}.', request_format: RUN_REQUEST_FORMAT, + response_format: RUN_RESPONSE_FORMAT, }, ); @@ -349,6 +428,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E description: 'Start a Pi turn and return immediately; watch agent::events (group_id = session_id) for progress and turn_end.', request_format: RUN_REQUEST_FORMAT, + response_format: START_RESPONSE_FORMAT, }, ); @@ -365,6 +445,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E description: 'Inject a steering instruction into a live Pi run; applied after the current tool calls finish.', request_format: STEER_REQUEST_FORMAT, + response_format: STEER_RESPONSE_FORMAT, }, ); @@ -381,6 +462,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E description: 'Queue a follow-up message for a live Pi run; processed after the agent would otherwise stop.', request_format: STEER_REQUEST_FORMAT, + response_format: FOLLOWUP_RESPONSE_FORMAT, }, ); @@ -396,6 +478,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E { description: 'Interrupt a live Pi run for a session.', request_format: SESSION_ID_FORMAT, + response_format: STOP_RESPONSE_FORMAT, }, ); @@ -409,12 +492,14 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E { description: 'Point-in-time status of a Pi session.', request_format: SESSION_ID_FORMAT, + response_format: STATUS_RESPONSE_FORMAT, }, ); iii.registerFunction('pi::sessions::list', async () => ({ sessions: await listSessions(iii) }), { description: 'List every Pi session this worker has run.', request_format: { type: 'object', properties: {} }, + response_format: SESSIONS_RESPONSE_FORMAT, }); iii.registerFunction( @@ -425,6 +510,7 @@ export function register(iii: ISdk, getCfg: () => Config, emit: Emit, emitRaw: E description: 'Alias for pi::run under the shared agent entrypoint: run a turn for {session_id, messages} and return when it ends.', request_format: RUN_REQUEST_FORMAT, + response_format: RUN_RESPONSE_FORMAT, }, ); }