reflectt · itskai-dev · May 4, 2026 · May 4, 2026
@@ -3,34 +3,30 @@
 
 // Command Drain — node-side cloud command queue runtime.
 //
-// Behavior-zero relocation of the inline command runtime previously
-// living in `src/cloud.ts`. Same poll cadence (10s active / 60s idle),
-// same dispatch (inline if/else by `cmd.type`), same ack contract.
-// No registry yet — that lands in step 2 of §11e in
-// `reflectt-cloud/docs/COMMAND_PLUGIN_SKILL_ARCHITECTURE.md`.
+// §11e step 1 (kai msg-1777492135899 / link msg-1777492193718) extracted
+// the inline poll/dispatch/ack from `src/cloud.ts` into this file with
+// inline if/else dispatch.
 //
-// Scope locks (kai msg-1777492135899 / link msg-1777492193718):
-//   - keep step 1 behavior-zero
+// §11e step 2 (kai msg-1777932028252 / link msg-1777932030625) replaces
+// that inline if/else with a registry lookup so adding a new verb is one
+// entry in `./registry.ts` plus one file in `./handlers/`. Behavior-zero
+// — same poll cadence, same ack contract, same unknown-skip semantics.
+//
+// Scope locks (still in force):
 //   - same poll cadence
 //   - same ack contract
-//   - same inline dispatch semantics (no registry creep until step 2)
 //   - no auth / intake churn
-//   - no new verbs
+//   - no new verbs (the two existing verbs are the first two registry entries)
 
-import { state, config, isIdle, cloudGet, cloudPost, markCloudActivity } from '../cloud.js'
+import { state, config, isIdle, cloudGet, cloudPost } from '../cloud.js'
+import { COMMAND_REGISTRY } from './registry.js'
+import type { PendingCommand } from './types.js'
 
 export const COMMAND_POLL_ACTIVE_MS = 10_000   // 10s when active
 export const COMMAND_POLL_IDLE_MS = 60_000     // 60s when idle
 let commandPollErrors = 0
 let lastCommandPollAt = 0
 
-interface PendingCommand {
-  id: string
-  type: string
-  payload: Record<string, unknown>
-  status: string
-}
-
 export async function pollAndProcessCommands(): Promise<void> {
   if (!state.hostId || !config || !state.running) return
 
@@ -71,131 +67,15 @@ export async function pollAndProcessCommands(): Promise<void> {
 }
 
 async function handleCommand(cmd: PendingCommand): Promise<void> {
-  if (cmd.type === 'context_sync') {
-    await handleContextSync(cmd)
-  } else if (cmd.type === 'run_approve') {
-    await handleRunApprove(cmd)
-  } else {
+  const handler = COMMAND_REGISTRY[cmd.type]
+  if (!handler) {
     console.log(`☁️  [Commands] Unknown command type: ${cmd.type} (${cmd.id}) — skipping`)
     // Ack unknown commands so they don't pile up
     await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
       action: 'complete',
       result: { skipped: true, reason: 'unknown_type' },
     })
-  }
-}
-
-async function handleRunApprove(cmd: PendingCommand): Promise<void> {
-  if (!state.hostId) return
-
-  const eventId = cmd.payload?.eventId as string
-  const decision = cmd.payload?.decision as string
-  const actor = cmd.payload?.actor as string || 'cloud-dashboard'
-  const rationale = cmd.payload?.rationale as string || ''
-
-  if (!eventId || !decision) {
-    console.warn(`☁️  [Commands] run_approve missing eventId/decision (${cmd.id}) — failing`)
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'fail',
-      error: 'eventId and decision are required',
-    })
     return
   }
-
-  console.log(`☁️  [Commands] Processing run_approve: ${decision} for ${eventId} (${cmd.id})`)
-
-  // Ack immediately
-  await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-    action: 'ack',
-  })
-
-  // Execute locally against the approval queue
-  const port = process.env.REFLECTT_NODE_PORT || '4445'
-  try {
-    const res = await fetch(`http://127.0.0.1:${port}/approval-queue/${encodeURIComponent(eventId)}/decide`, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify({ decision, actor, rationale }),
-    })
-
-    const result = await res.json().catch(() => ({ success: false }))
-
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'complete',
-      result: { eventId, decision, status: res.status, ...(result as Record<string, unknown>) },
-    })
-
-    console.log(`☁️  [Commands] run_approve ${decision} for ${eventId} — ${res.status}`)
-  } catch (err: any) {
-    console.warn(`☁️  [Commands] run_approve failed for ${eventId}: ${err?.message}`)
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'fail',
-      error: err?.message || 'Local approval-queue call failed',
-    })
-  }
-}
-
-async function handleContextSync(cmd: PendingCommand): Promise<void> {
-  if (!state.hostId) return
-
-  // Require explicit agent — no hardcoded fallback
-  const agent = (cmd.payload?.agent as string)?.trim()
-  if (!agent) {
-    console.warn(`☁️  [Commands] context_sync missing payload.agent (${cmd.id}) — failing`)
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'fail',
-      error: 'payload.agent is required',
-    })
-    return
-  }
-
-  console.log(`☁️  [Commands] Processing context_sync for agent=${agent} (${cmd.id})`)
-
-  // Ack immediately (in-progress)
-  await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-    action: 'ack',
-  })
-
-  // Fetch context snapshot from local node
-  const port = process.env.REFLECTT_NODE_PORT || '4445'
-  let contextData: Record<string, unknown>
-  try {
-    const localRes = await fetch(`http://127.0.0.1:${port}/context/inject/${encodeURIComponent(agent)}`)
-    if (!localRes.ok) throw new Error(`Local context fetch failed: ${localRes.status}`)
-    contextData = await localRes.json() as Record<string, unknown>
-  } catch (err: any) {
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'fail',
-      error: `Failed to fetch local context: ${err?.message}`,
-    })
-    throw err
-  }
-
-  // Push to cloud — use computed_at from injection payload when available
-  const computedAt = (typeof contextData.computed_at === 'number' && contextData.computed_at > 0)
-    ? contextData.computed_at
-    : Date.now()
-
-  const syncResult = await cloudPost(`/api/hosts/${state.hostId}/context/sync`, {
-    agent,
-    computed_at: computedAt,
-    budgets: contextData.budgets || { totalTokens: 0, layers: {} },
-    autosummary_enabled: Boolean(contextData.autosummary_enabled),
-    layers: contextData.layers || {},
-  })
-
-  if (syncResult.success) {
-    console.log(`☁️  [Commands] context_sync completed for ${agent} (${cmd.id})`)
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'complete',
-      result: { syncedAt: Date.now(), agent },
-    })
-    markCloudActivity() // Mark as active
-  } else {
-    console.warn(`☁️  [Commands] context_sync failed for ${agent}: ${syncResult.error}`)
-    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
-      action: 'fail',
-      error: syncResult.error,
-    })
-  }
+  await handler(cmd, { hostId: state.hostId! })
 }
@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) Reflectt AI
+
+// `context_sync` — fetches the local agent context snapshot and pushes
+// it to the cloud's `/context/sync` endpoint. Behavior-zero relocation
+// from `drain.ts` per §11e step 2 (kai msg-1777932028252 / link
+// msg-1777932030625).
+
+import { state, cloudPost, markCloudActivity } from '../../cloud.js'
+import type { CommandHandler } from '../types.js'
+
+export const handleContextSync: CommandHandler = async (cmd) => {
+  if (!state.hostId) return
+
+  // Require explicit agent — no hardcoded fallback
+  const agent = (cmd.payload?.agent as string)?.trim()
+  if (!agent) {
+    console.warn(`☁️  [Commands] context_sync missing payload.agent (${cmd.id}) — failing`)
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'fail',
+      error: 'payload.agent is required',
+    })
+    return
+  }
+
+  console.log(`☁️  [Commands] Processing context_sync for agent=${agent} (${cmd.id})`)
+
+  // Ack immediately (in-progress)
+  await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+    action: 'ack',
+  })
+
+  // Fetch context snapshot from local node
+  const port = process.env.REFLECTT_NODE_PORT || '4445'
+  let contextData: Record<string, unknown>
+  try {
+    const localRes = await fetch(`http://127.0.0.1:${port}/context/inject/${encodeURIComponent(agent)}`)
+    if (!localRes.ok) throw new Error(`Local context fetch failed: ${localRes.status}`)
+    contextData = await localRes.json() as Record<string, unknown>
+  } catch (err: any) {
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'fail',
+      error: `Failed to fetch local context: ${err?.message}`,
+    })
+    throw err
+  }
+
+  // Push to cloud — use computed_at from injection payload when available
+  const computedAt = (typeof contextData.computed_at === 'number' && contextData.computed_at > 0)
+    ? contextData.computed_at
+    : Date.now()
+
+  const syncResult = await cloudPost(`/api/hosts/${state.hostId}/context/sync`, {
+    agent,
+    computed_at: computedAt,
+    budgets: contextData.budgets || { totalTokens: 0, layers: {} },
+    autosummary_enabled: Boolean(contextData.autosummary_enabled),
+    layers: contextData.layers || {},
+  })
+
+  if (syncResult.success) {
+    console.log(`☁️  [Commands] context_sync completed for ${agent} (${cmd.id})`)
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'complete',
+      result: { syncedAt: Date.now(), agent },
+    })
+    markCloudActivity() // Mark as active
+  } else {
+    console.warn(`☁️  [Commands] context_sync failed for ${agent}: ${syncResult.error}`)
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'fail',
+      error: syncResult.error,
+    })
+  }
+}
@@ -0,0 +1,59 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) Reflectt AI
+
+// `run_approve` — relays a cloud-side approval decision into the local
+// approval queue. Behavior-zero relocation from `drain.ts` per §11e
+// step 2 (kai msg-1777932028252 / link msg-1777932030625).
+
+import { state, cloudPost } from '../../cloud.js'
+import type { CommandHandler } from '../types.js'
+
+export const handleRunApprove: CommandHandler = async (cmd) => {
+  if (!state.hostId) return
+
+  const eventId = cmd.payload?.eventId as string
+  const decision = cmd.payload?.decision as string
+  const actor = cmd.payload?.actor as string || 'cloud-dashboard'
+  const rationale = cmd.payload?.rationale as string || ''
+
+  if (!eventId || !decision) {
+    console.warn(`☁️  [Commands] run_approve missing eventId/decision (${cmd.id}) — failing`)
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'fail',
+      error: 'eventId and decision are required',
+    })
+    return
+  }
+
+  console.log(`☁️  [Commands] Processing run_approve: ${decision} for ${eventId} (${cmd.id})`)
+
+  // Ack immediately
+  await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+    action: 'ack',
+  })
+
+  // Execute locally against the approval queue
+  const port = process.env.REFLECTT_NODE_PORT || '4445'
+  try {
+    const res = await fetch(`http://127.0.0.1:${port}/approval-queue/${encodeURIComponent(eventId)}/decide`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ decision, actor, rationale }),
+    })
+
+    const result = await res.json().catch(() => ({ success: false }))
+
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'complete',
+      result: { eventId, decision, status: res.status, ...(result as Record<string, unknown>) },
+    })
+
+    console.log(`☁️  [Commands] run_approve ${decision} for ${eventId} — ${res.status}`)
+  } catch (err: any) {
+    console.warn(`☁️  [Commands] run_approve failed for ${eventId}: ${err?.message}`)
+    await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
+      action: 'fail',
+      error: err?.message || 'Local approval-queue call failed',
+    })
+  }
+}
@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) Reflectt AI
+
+// Command registry — the single dispatch lookup `drain.ts` uses to map
+// `cmd.type` → handler. §11e step 2 of
+// `reflectt-cloud/docs/COMMAND_PLUGIN_SKILL_ARCHITECTURE.md`.
+//
+// Locks (kai msg-1777932028252 / link msg-1777932030625):
+//   - keep step 2 behavior-zero
+//   - no new verbs (the two existing verbs become the first two registry
+//     entries — no scope creep)
+//   - drain.ts dispatches via this registry; unknown `cmd.type` falls
+//     back to the existing skip-and-ack path in drain
+//
+// Adding a new verb after this lane closes is one entry here plus one
+// per-verb file in `./handlers/` — no ad-hoc dispatch edits in drain.
+
+import { handleContextSync } from './handlers/context-sync.js'
+import { handleRunApprove } from './handlers/run-approve.js'
+import type { CommandHandler } from './types.js'
+
+export const COMMAND_REGISTRY: Record<string, CommandHandler> = {
+  context_sync: handleContextSync,
+  run_approve: handleRunApprove,
+}
@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) Reflectt AI
+
+// Command runtime — shared types.
+//
+// §11e step 2 of `reflectt-cloud/docs/COMMAND_PLUGIN_SKILL_ARCHITECTURE.md`.
+// Locks (kai msg-1777932028252 / link msg-1777932030625):
+//   - shared handler/context types
+//   - behavior-zero
+//   - no new verbs
+
+/** A pending command pulled off the cloud's command queue (one entry of
+ *  `GET /api/hosts/:hostId/commands?status=pending`). */
+export interface PendingCommand {
+  id: string
+  type: string
+  payload: Record<string, unknown>
+  status: string
+}
+
+/** Per-dispatch context passed by `drain.ts` into the handler. Carries
+ *  what every handler needs at the boundary today (the validated host id).
+ *  Future verbs may extend this — kept narrow on purpose. */
+export interface CommandContext {
+  hostId: string
+}
+
+/** Async handler for a single command verb. Errors thrown out of the
+ *  handler are caught by `drain.ts` and acked as `fail` on the cloud
+ *  queue — same contract step 1 already implemented inline. */
+export type CommandHandler = (cmd: PendingCommand, ctx: CommandContext) => Promise<void>