Skip to content

refactor(commands): extract command poll + dispatch + ack into commands/drain.ts #1317

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
itskai-dev merged 1 commit into from
Apr 29, 2026
Merged

refactor(commands): extract command poll + dispatch + ack into commands/drain.ts #1317

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 12 additions & 185 deletions src/cloud.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import { getRequestMetrics } from './request-tracker.js'
import { listApprovalQueue, listAgentEvents, listAgentRuns, type AgentRun } from './agent-runs.js'
import { syncTeamContextToAgents } from './team-context-writer.js'
import { getUnpushedTrustEvents, markTrustEventsPushed } from './trust-events.js'
import { pollAndProcessCommands, COMMAND_POLL_ACTIVE_MS, COMMAND_POLL_IDLE_MS } from './commands/drain.js'

/**
* Docker identity guard: detect when a container has inherited cloud
Expand Down Expand Up @@ -169,7 +170,7 @@ export function queueCanvasPushEvent(event: Record<string, unknown>): void {
}

/** Check if the system is idle */
function isIdle(): boolean {
export function isIdle(): boolean {
return Date.now() - lastActivityAt > IDLE_THRESHOLD_MS
}

Expand Down Expand Up @@ -227,8 +228,8 @@ export function getConnectionHealth() {
}
}

let config: CloudConfig | null = null
let state: CloudState = {
export let config: CloudConfig | null = null
export let state: CloudState = {
hostId: null,
credential: null,
heartbeatTimer: null,
Expand Down Expand Up @@ -2045,18 +2046,10 @@ async function syncInsightsToCloud(): Promise<void> {
}

// ---- Command polling + context_sync handler ----

const COMMAND_POLL_ACTIVE_MS = 10_000 // 10s when active
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
}
//
// Relocated to src/commands/drain.ts (§11e step 1, behavior-zero).
// `pollAndProcessCommands` is imported from there at the top of this
// file; the in-flight call sites below are unchanged.

// ─── Capability context injection ─────────────────────────────────────────────

Expand Down Expand Up @@ -2138,174 +2131,8 @@ async function syncCapabilityContext(): Promise<void> {
}
}

async function pollAndProcessCommands(): Promise<void> {
if (!state.hostId || !config || !state.running) return

const now = Date.now()
const interval = isIdle() ? COMMAND_POLL_IDLE_MS : COMMAND_POLL_ACTIVE_MS
if (now - lastCommandPollAt < interval) return
lastCommandPollAt = now

const result = await cloudGet<{ commands: PendingCommand[] }>(
`/api/hosts/${state.hostId}/commands?status=pending`
)

if (!result.success || !result.data?.commands) {
commandPollErrors++
if (commandPollErrors <= 3 || commandPollErrors % 20 === 0) {
console.warn(`☁️ [Commands] Poll failed (${commandPollErrors}): ${result.error}`)
}
return
}

if (commandPollErrors > 0) {
console.log(`☁️ [Commands] Poll recovered after ${commandPollErrors} errors`)
commandPollErrors = 0
}

for (const cmd of result.data.commands) {
try {
await handleCommand(cmd)
} catch (err: any) {
console.warn(`☁️ [Commands] Failed to handle ${cmd.type} (${cmd.id}): ${err?.message}`)
// Ack as failed so it doesn't re-run
await cloudPost(`/api/hosts/${state.hostId}/commands/${cmd.id}/ack`, {
action: 'fail',
error: err?.message || 'Handler error',
}).catch(() => {})
}
}
}

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 {
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,
})
}
}
// pollAndProcessCommands / handleCommand / handleContextSync /
// handleRunApprove relocated to src/commands/drain.ts (§11e step 1).

// ---- HTTP helper ----

Expand Down Expand Up @@ -2401,7 +2228,7 @@ async function attemptCredentialReclaim(): Promise<boolean> {
}
}

async function cloudGet<T = unknown>(path: string, _retried = false): Promise<CloudApiResponse<T>> {
export async function cloudGet<T = unknown>(path: string, _retried = false): Promise<CloudApiResponse<T>> {
if (!config) return { success: false, error: 'Not configured' }

try {
Expand Down Expand Up @@ -2435,7 +2262,7 @@ async function cloudGet<T = unknown>(path: string, _retried = false): Promise<Cl
}
}

async function cloudPost<T = unknown>(path: string, body: unknown, _retried = false): Promise<CloudApiResponse<T>> {
export async function cloudPost<T = unknown>(path: string, body: unknown, _retried = false): Promise<CloudApiResponse<T>> {
if (!config) return { success: false, error: 'Not configured' }

try {
Expand Down
Loading
Loading