feat: dual-lane queue for concurrent message and task execution

src/backends/daytona-backend.ts

@@ -479,11 +479,11 @@ export class DaytonaBackend implements AgentBackend {
     return true;
   }
 
-  closeStdin(groupFolder: string): void {
+  closeStdin(groupFolder: string, inputSubdir: string = 'input'): void {
     const meta = this.sandboxes.get(groupFolder);
     if (!meta) return;
 
-    uploadFile(meta.sandbox, 'workspace/ipc/input/_close', '').catch((err) => {
+    uploadFile(meta.sandbox, `workspace/ipc/${inputSubdir}/_close`, '').catch((err) => {
       logger.warn({ groupFolder, error: err }, 'Failed to write close sentinel to Daytona sandbox');
     });
   }

src/backends/hetzner-backend.ts

@@ -280,7 +280,7 @@ runcmd:
     return true;
   }
 
-  closeStdin(groupFolder: string): void {
+  closeStdin(groupFolder: string, _inputSubdir?: string): void {
     if (!this.s3) return;
 
     const messageId = `${Date.now()}-${crypto.randomUUID().slice(0, 8)}`;

src/backends/local-backend.ts

@@ -47,6 +47,7 @@ function getHomeDir(): string {
 function buildVolumeMounts(
   group: AgentOrGroup,
   isMain: boolean,
+  isScheduledTask: boolean = false,
 ): VolumeMount[] {
   const mounts: VolumeMount[] = [];
   const homeDir = getHomeDir();
@@ -140,11 +141,46 @@ function buildVolumeMounts(
   fs.mkdirSync(path.join(groupIpcDir, 'messages'), { recursive: true });
   fs.mkdirSync(path.join(groupIpcDir, 'tasks'), { recursive: true });
   fs.mkdirSync(path.join(groupIpcDir, 'input'), { recursive: true });
-  mounts.push({
-    hostPath: groupIpcDir,
-    containerPath: '/workspace/ipc',
-    readonly: false,
-  });
+  fs.mkdirSync(path.join(groupIpcDir, 'input-task'), { recursive: true });
+
+  if (isScheduledTask) {
+    // Task containers get input-task/ mounted as their /workspace/ipc/input
+    // so follow-up messages don't cross lanes
+    mounts.push({
+      hostPath: path.join(groupIpcDir, 'messages'),
+      containerPath: '/workspace/ipc/messages',
+      readonly: false,
+    });
+    mounts.push({
+      hostPath: path.join(groupIpcDir, 'tasks'),
+      containerPath: '/workspace/ipc/tasks',
+      readonly: false,
+    });
+    mounts.push({
+      hostPath: path.join(groupIpcDir, 'input-task'),
+      containerPath: '/workspace/ipc/input',
+      readonly: false,
+    });
+    // Mount IPC root for agent_registry.json and other top-level files
+    // Use a separate mount point so agent-runner can still find them
+    const ipcFiles = ['agent_registry.json', 'tasks.json', 'groups.json'];
+    for (const file of ipcFiles) {
+      const filePath = path.join(groupIpcDir, file);
+      if (fs.existsSync(filePath)) {
+        mounts.push({
+          hostPath: filePath,
+          containerPath: `/workspace/ipc/${file}`,
+          readonly: true,
+        });
+      }
+    }
+  } else {
+    mounts.push({
+      hostPath: groupIpcDir,
+      containerPath: '/workspace/ipc',
+      readonly: false,
+    });
+  }
 
   // Environment file
   const envDir = path.join(DATA_DIR, 'env');
@@ -229,7 +265,7 @@ export class LocalBackend implements AgentBackend {
     const groupDir = path.join(GROUPS_DIR, folder);
     fs.mkdirSync(groupDir, { recursive: true });
 
-    const mounts = buildVolumeMounts(group, input.isMain);
+    const mounts = buildVolumeMounts(group, input.isMain, input.isScheduledTask);
     const safeName = folder.replace(/[^a-zA-Z0-9-]/g, '-');
     const containerName = `nanoclaw-${safeName}-${Date.now()}`;
     const containerArgs = buildContainerArgs(mounts, containerName);
@@ -515,8 +551,8 @@ export class LocalBackend implements AgentBackend {
     }
   }
 
-  closeStdin(groupFolder: string): void {
-    const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, 'input');
+  closeStdin(groupFolder: string, inputSubdir: string = 'input'): void {
+    const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, inputSubdir);
     try {
       fs.mkdirSync(inputDir, { recursive: true });
       fs.writeFileSync(path.join(inputDir, '_close'), '');

src/backends/railway-backend.ts

@@ -155,7 +155,7 @@ export class RailwayBackend implements AgentBackend {
     return true;
   }
 
-  closeStdin(groupFolder: string): void {
+  closeStdin(groupFolder: string, _inputSubdir?: string): void {
     if (!this.s3) return;
 
     const messageId = `${Date.now()}-${crypto.randomUUID().slice(0, 8)}`;

src/backends/sprites-backend.ts

@@ -501,9 +501,9 @@ export class SpritesBackend implements AgentBackend {
     return true;
   }
 
-  closeStdin(groupFolder: string): void {
+  closeStdin(groupFolder: string, inputSubdir: string = 'input'): void {
     const sprite = this.getSpriteClient(groupFolder);
-    sprite.writeFile('/workspace/ipc/input/_close', '').catch((err) => {
+    sprite.writeFile(`/workspace/ipc/${inputSubdir}/_close`, '').catch((err) => {
       logger.warn({ groupFolder, error: err }, 'Failed to write close sentinel to Sprite');
     });
   }

src/backends/types.ts

@@ -89,8 +89,8 @@ export interface AgentBackend {
   /** Send a follow-up message to an active agent via IPC. Returns true if sent. */
   sendMessage(groupFolder: string, text: string): boolean;
 
-  /** Signal an active agent to wind down. */
-  closeStdin(groupFolder: string): void;
+  /** Signal an active agent to wind down. Optional inputSubdir for task lane isolation. */
+  closeStdin(groupFolder: string, inputSubdir?: string): void;
 
   /** Write IPC data files (tasks snapshot, groups snapshot, agent registry). */
   writeIpcData(groupFolder: string, filename: string, data: string): void;

src/config.ts

@@ -54,6 +54,10 @@ export const MAX_CONCURRENT_CONTAINERS = Math.max(
   1,
   parseInt(process.env.MAX_CONCURRENT_CONTAINERS || '5', 10) || 5,
 );
+export const MAX_TASK_CONTAINERS = Math.max(
+  1,
+  parseInt(process.env.MAX_TASK_CONTAINERS || String(MAX_CONCURRENT_CONTAINERS - 1), 10),
+);
 
 function escapeRegex(str: string): string {
   return str.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');

src/db.ts

@@ -475,6 +475,13 @@ export function getDueTasks(): ScheduledTask[] {
     .all(now) as ScheduledTask[];
 }
 
+/** Advance next_run without touching last_run/last_result (used before enqueue). */
+export function advanceTaskNextRun(id: string, nextRun: string | null): void {
+  db.query(
+    `UPDATE scheduled_tasks SET next_run = ?, status = CASE WHEN ? IS NULL THEN 'completed' ELSE status END WHERE id = ?`,
+  ).run(nextRun, nextRun, id);
+}
+
 export function updateTaskAfterRun(
   id: string,
   nextRun: string | null,