Plan 323 P3 + P3b — models on this computer, diagnosed, managed, and changed by Helmsman with approval

apps/aflow-executor-ai/src/handlers/ai/handlers/agentTurnModel.ts

@@ -10,6 +10,7 @@ import type { AgentTurnInput } from '../schema.js';
 import type { HandlerDeps } from './types.js';
 import { chatMessageToAiMessage } from './agentMessageConversion.js';
 import { toolImageResolver } from './mediaSourceRef.js';
+import { reportRuntimePhases } from './runtimePhaseStatus.js';
 import {
   buildFunctionDeclarations,
   mapToolCallsToDecision,
@@ -438,6 +439,7 @@ async function streamAndCollect(
     attempt: ctx.job.attempt,
     signal: ctx.signal,
     onStreamProgress: ctx.reportProgress,
+    onRuntimePhase: reportRuntimePhases(ctx, client.getModel(p.model)?.displayName ?? p.model),
     maxRetries: 0, // Agent turns are interactive — let orchestrator handle retries with user feedback
   });
 
@@ -741,6 +743,7 @@ async function runGenerateJsonPath(
     attempt: ctx.job.attempt,
     signal: ctx.signal,
     onStreamProgress: ctx.reportProgress,
+    onRuntimePhase: reportRuntimePhases(ctx, modelDef?.displayName ?? model),
     maxRetries: 0,
   });
 

apps/aflow-executor-ai/src/handlers/ai/handlers/runtimePhaseStatus.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it, vi } from 'vitest';
+import type { ExecutorContext } from '@aflow/executor-runtime';
+import { reportRuntimePhases } from './runtimePhaseStatus.js';
+
+function context() {
+  const emitLiveDelta = vi.fn(async () => undefined);
+  const reportProgress = vi.fn();
+  const ctx = { emitLiveDelta, reportProgress, log: { warn: vi.fn() } };
+  return { ctx: ctx as unknown as ExecutorContext, emitLiveDelta, reportProgress };
+}
+
+describe('a local model’s silent stretches', () => {
+  it('are one line each on the status channel, naming the model that loads', () => {
+    const { ctx, emitLiveDelta } = context();
+    const report = reportRuntimePhases(ctx, 'Gemma 4 26B (this computer)');
+
+    report({ phase: 'loading_model' });
+    report({ phase: 'reading_prompt' });
+
+    expect(emitLiveDelta.mock.calls).toEqual([
+      ['status', 'Loading Gemma 4 26B (this computer) into memory…\n'],
+      ['status', 'Reading the conversation…\n'],
+    ]);
+  });
+
+  it('count as progress, so a long load is not cut as a stall', () => {
+    const { ctx, reportProgress } = context();
+
+    reportRuntimePhases(ctx, 'm')({ phase: 'loading_model' });
+
+    expect(reportProgress).toHaveBeenCalledTimes(1);
+  });
+});

apps/aflow-executor-ai/src/handlers/ai/handlers/runtimePhaseStatus.ts

@@ -0,0 +1,31 @@
+import type { ExecutorContext } from '@aflow/executor-runtime';
+import type { RuntimePhase } from '@aflow/ai-client';
+
+export function runtimePhaseLine(phase: RuntimePhase, modelName: string): string {
+  switch (phase.phase) {
+    case 'loading_model':
+      return `Loading ${modelName} into memory…`;
+    case 'reading_prompt':
+      return 'Reading the conversation…';
+  }
+}
+
+/**
+ * Puts a local model's silent stretches on the step's `status` channel, where
+ * the chat shows them in place of a bare spinner until the first token. Each
+ * phase also slides the idle window: a load is the runtime working, not a
+ * stall.
+ */
+export function reportRuntimePhases(
+  ctx: ExecutorContext,
+  modelName: string,
+): (phase: RuntimePhase) => void {
+  return (phase) => {
+    ctx.reportProgress?.();
+    ctx.emitLiveDelta('status', `${runtimePhaseLine(phase, modelName)}\n`).catch((err: unknown) => {
+      ctx.log.warn('Live status delta dropped', {
+        error: err instanceof Error ? err.message : String(err),
+      });
+    });
+  };
+}

apps/aflow-executor-host/src/__tests__/machine.test.ts

@@ -21,16 +21,17 @@ describe('createMachineReader', () => {
       totalMemoryBytes: 24 * GIB,
       osRelease: '25.3.0',
       run,
+      freeDisk: async () => undefined,
     };
 
     const first = createMachineReader(options);
-    expect(first.current()).toEqual({ totalMemoryBytes: 24 * GIB, unifiedMemory: true });
+    expect(await first.current()).toEqual({ totalMemoryBytes: 24 * GIB, unifiedMemory: true });
     await first.start();
-    expect(first.current().gpuWorkingSetBytes).toBe(19_069_665_280);
+    expect((await first.current()).gpuWorkingSetBytes).toBe(19_069_665_280);
 
     const second = createMachineReader(options);
     await second.start();
-    expect(second.current().gpuWorkingSetBytes).toBe(19_069_665_280);
+    expect((await second.current()).gpuWorkingSetBytes).toBe(19_069_665_280);
     expect(run).toHaveBeenCalledTimes(1);
   });
 
@@ -64,9 +65,10 @@ describe('createMachineReader', () => {
       totalMemoryBytes: 64 * GIB,
       osRelease: '6.8',
       run: async () => '24576\n49152\n',
+      freeDisk: async () => undefined,
     });
     await nvidia.start();
-    expect(nvidia.current()).toEqual({
+    expect(await nvidia.current()).toEqual({
       totalMemoryBytes: 64 * GIB,
       unifiedMemory: false,
       gpuWorkingSetBytes: 49152 * 1024 ** 2,
@@ -81,8 +83,32 @@ describe('createMachineReader', () => {
       run: async () => {
         throw new Error('nvidia-smi: command not found');
       },
+      freeDisk: async () => undefined,
     });
     await none.start();
-    expect(none.current().gpuWorkingSetBytes).toBeUndefined();
+    expect((await none.current()).gpuWorkingSetBytes).toBeUndefined();
+  });
+
+  it('reads the free disk under the model store at every read, since downloads change it', async () => {
+    let free = 100 * GIB;
+    const paths: string[] = [];
+    const reader = createMachineReader({
+      hostDir: await hostDir(),
+      platform: 'linux',
+      arch: 'x64',
+      totalMemoryBytes: 16 * GIB,
+      osRelease: '6.8',
+      run: async () => '',
+      modelStore: '/models',
+      freeDisk: async (path) => {
+        paths.push(path);
+        return free;
+      },
+    });
+
+    expect((await reader.current()).freeDiskBytes).toBe(100 * GIB);
+    free = 84 * GIB;
+    expect((await reader.current()).freeDiskBytes).toBe(84 * GIB);
+    expect(paths).toEqual(['/models', '/models']);
   });
 });

apps/aflow-executor-host/src/browser/approvalStore.ts

@@ -3,7 +3,7 @@ import {
   getWriteApprovalGrant,
   readBrowserAsk,
   rememberBrowserAsk,
-  spendBrowserApproval,
+  spendWriteApprovalGrant,
 } from '@aflow/redis';
 
 import type { ApprovalStore } from './actionApproval.js';
@@ -14,7 +14,7 @@ export function redisApprovalStore(redis: Redis): ApprovalStore {
     grant: async (scope, requestHash) =>
       await getWriteApprovalGrant(redis, scope.tenantId, scope.runId, requestHash),
     spend: async (scope, grant) =>
-      await spendBrowserApproval(redis, scope.tenantId, scope.runId, grant),
+      await spendWriteApprovalGrant(redis, scope.tenantId, scope.runId, grant),
     recall: async (scope, callKey) =>
       await readBrowserAsk(redis, scope.tenantId, scope.runId, callKey),
     remember: async (scope, callKey, requestHash) => {

apps/aflow-executor-host/src/index.ts

@@ -339,7 +339,7 @@ async function main(): Promise<void> {
     const inventory: HostInventory = {
       hostname,
       observedAt: new Date().toISOString(),
-      machine: machine.current(),
+      machine: await machine.current(),
       runtimes,
       harnesses,
       maxConcurrentHarnessRuns,

apps/aflow-executor-host/src/machine.ts

@@ -12,7 +12,7 @@
  * says it does not know rather than guessing.
  */
 import { execFile } from 'node:child_process';
-import { readFile, writeFile } from 'node:fs/promises';
+import { readFile, statfs, writeFile } from 'node:fs/promises';
 import os from 'node:os';
 import { join } from 'node:path';
 import type { HostMachine } from '@aflow/redis';
@@ -62,9 +62,30 @@ async function readGpuWorkingSet(platform: string, run: RunCommand): Promise<num
   return mib.length > 0 ? Math.max(...mib) * 1024 ** 2 : undefined;
 }
 
+/**
+ * Where Ollama keeps its downloads: `OLLAMA_MODELS` when the executor's
+ * environment names it, the runtime's default otherwise. A runtime started
+ * by a service manager with another `OLLAMA_MODELS` keeps its downloads on a
+ * disk this cannot see, which only an operator who moved them has.
+ */
+function defaultModelStore(): string {
+  return process.env['OLLAMA_MODELS'] ?? join(os.homedir(), '.ollama', 'models');
+}
+
+export type ReadFreeDisk = (path: string) => Promise<number | undefined>;
+
+const readFreeDisk: ReadFreeDisk = async (path) => {
+  // The store may not exist before the first pull; its disk is the home's.
+  for (const candidate of [path, os.homedir()]) {
+    const stats = await statfs(candidate).catch(() => undefined);
+    if (stats) return stats.bavail * stats.bsize;
+  }
+  return undefined;
+};
+
 export interface MachineReader {
   /** The machine as read so far; the GPU figure arrives once the probe has run. */
-  current(): HostMachine;
+  current(): Promise<HostMachine>;
   /** Reads the GPU figure, from the cache when it still describes this machine. */
   start(): Promise<void>;
 }
@@ -76,19 +97,26 @@ export function createMachineReader(options: {
   totalMemoryBytes?: number;
   osRelease?: string;
   run?: RunCommand;
+  modelStore?: string;
+  freeDisk?: ReadFreeDisk;
 }): MachineReader {
   const platform = options.platform ?? process.platform;
   const totalMemoryBytes = options.totalMemoryBytes ?? os.totalmem();
   const osRelease = options.osRelease ?? os.release();
   const run = options.run ?? runCommand;
   const cachePath = join(options.hostDir, CACHE_FILE);
+  const modelStore = options.modelStore ?? defaultModelStore();
+  const freeDisk = options.freeDisk ?? readFreeDisk;
   const machine: HostMachine = {
     totalMemoryBytes,
     unifiedMemory: platform === 'darwin' && (options.arch ?? process.arch) === 'arm64',
   };
 
   return {
-    current: () => ({ ...machine }),
+    async current() {
+      const freeDiskBytes = await freeDisk(modelStore).catch(() => undefined);
+      return { ...machine, ...(freeDiskBytes !== undefined ? { freeDiskBytes } : {}) };
+    },
     async start() {
       const cached = await readFile(cachePath, 'utf8')
         .then((text) => JSON.parse(text) as CachedReading)

apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/dispatchInlineOp.ts

@@ -32,6 +32,7 @@ import {
   handleWebhookCrudInline,
   handleWorkflowCrudInline,
   handlePlanNodeInline,
+  handleModelRuntimeInline,
   handleCoachCrudInline,
   handleProposalCrudInline,
   handleCatalogAgentInline,
@@ -497,6 +498,20 @@ export async function dispatchInlineOp(
       ...(parentStepExecutionId ? { parentStepExecutionId } : {}),
       ...workflowExecutionFields,
     });
+  } else if (stepDef.operation.startsWith('model.')) {
+    await handleModelRuntimeInline({
+      redis,
+      payloadStore,
+      context,
+      stepDef,
+      stepExecutionId,
+      idempotencyKey,
+      resolvedInputRef,
+      attempt,
+      scheduledAtMs: now,
+      ...(parentStepExecutionId ? { parentStepExecutionId } : {}),
+      ...workflowExecutionFields,
+    });
   } else if (stepDef.operation.startsWith('catalog.agent.')) {
     await handleCatalogAgentInline({
       redis,

apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/index.ts

@@ -14,6 +14,7 @@ export { handleScheduleSnoozeInline } from './scheduleSnooze.js';
 export { handleWebhookCrudInline } from './webhookCrud.js';
 export { handleWorkflowCrudInline } from './workflowCrud/index.js';
 export { handlePlanNodeInline } from './plan/index.js';
+export { handleModelRuntimeInline } from './model/index.js';
 export { handleCoachCrudInline } from './coachCrud.js';
 export { handleProposalCrudInline } from './proposalCrud.js';
 export { handleCatalogAgentInline } from './catalogAgent.js';