fix(pi): publish typed response schemas and strip the $schema meta-ref

pi/src/config.ts

@@ -30,9 +30,12 @@ export type Config = z.infer<typeof ConfigSchema>;
 export const RuntimeConfigSchema = ConfigSchema.omit({ engine_url: true });
 export type RuntimeConfig = z.infer<typeof RuntimeConfigSchema>;
 
-/** 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<string, unknown> {
-  return z.toJSONSchema(RuntimeConfigSchema) as Record<string, unknown>;
+  const out = z.toJSONSchema(RuntimeConfigSchema) as Record<string, unknown>;
+  delete out.$schema;
+  return out;
 }
 
 /** The runtime slice of a full config, for use as `initial_value`. */

pi/src/configuration.ts

@@ -62,10 +62,18 @@ export async function fetchRuntime(iii: ISdk): Promise<RuntimeConfig | null> {
  */
 export async function bindConfigTrigger(iii: ISdk, onChange: () => Promise<void>): Promise<void> {
   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,

pi/src/run.ts

@@ -84,9 +84,87 @@
   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<string, unknown> {
+  const out = z.toJSONSchema(schema) as Record<string, unknown>;
+  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<string, LiveRun>();
@@ -318,6 +396,7 @@
       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 @@
       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 @@
       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 @@
       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 @@
     {
       description: 'Interrupt a live Pi run for a session.',
       request_format: SESSION_ID_FORMAT,
+      response_format: STOP_RESPONSE_FORMAT,
     },
   );
 
@@ -409,12 +492,14 @@
     {
       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 @@
       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,
     },
   );
 }