diff --git a/src/commands/drain.ts b/src/commands/drain.ts index 75a5d671..593e61f6 100644 --- a/src/commands/drain.ts +++ b/src/commands/drain.ts @@ -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 - status: string -} - export async function pollAndProcessCommands(): Promise { if (!state.hostId || !config || !state.running) return @@ -71,131 +67,15 @@ export async function pollAndProcessCommands(): Promise { } async function handleCommand(cmd: PendingCommand): Promise { - 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 { - 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) }, - }) - - 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 { - 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 - 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 - } 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! }) } diff --git a/src/commands/handlers/context-sync.ts b/src/commands/handlers/context-sync.ts new file mode 100644 index 00000000..e83b3a3f --- /dev/null +++ b/src/commands/handlers/context-sync.ts @@ -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 + 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 + } 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, + }) + } +} diff --git a/src/commands/handlers/run-approve.ts b/src/commands/handlers/run-approve.ts new file mode 100644 index 00000000..b28ffa4a --- /dev/null +++ b/src/commands/handlers/run-approve.ts @@ -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) }, + }) + + 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', + }) + } +} diff --git a/src/commands/registry.ts b/src/commands/registry.ts new file mode 100644 index 00000000..799aeb33 --- /dev/null +++ b/src/commands/registry.ts @@ -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 = { + context_sync: handleContextSync, + run_approve: handleRunApprove, +} diff --git a/src/commands/types.ts b/src/commands/types.ts new file mode 100644 index 00000000..3268fef0 --- /dev/null +++ b/src/commands/types.ts @@ -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 + 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