Add an ncl tasks resource so operators can stop runaway tasks

Adds a host-side `ncl tasks` CLI resource (list/get/cancel/pause/resume)
backed by per-session inbound.db rows rather than the central DB, giving
operators a way to stop a runaway scheduled task without asking the agent
that owns it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.claude/skills/debug/SKILL.md

@@ -206,6 +206,29 @@ ncl groups restart --id <group-id> --message "on_wake test"
 
 Without `--message`, the container comes back on the next user message. From inside a container, `--id` is auto-filled and only the calling session restarts.
 
+## Runaway task
+
+A scheduled task (cron) that keeps firing is a `messages_in` row with `kind='task'` in a session's `inbound.db`. When an agent won't stop one — or you'd rather not ask the misbehaving agent — use the `ncl tasks` operator surface to inspect and stop it directly from the host:
+
+```bash
+# List every pending/paused task across all groups (one row per series).
+ncl tasks list
+
+# Scope to a single agent group.
+ncl tasks list --group <group-id>
+
+# Inspect one task by id or series id.
+ncl tasks get --id <task-id>
+
+# Stop it. cancel/pause/resume match by id OR series id, so a recurring
+# task's live next occurrence is caught, not just the row you looked up.
+ncl tasks cancel --id <task-id>
+ncl tasks pause  --id <task-id>
+ncl tasks resume --id <task-id>
+```
+
+The host is the legitimate writer of `inbound.db`, so these apply straight to the session DB without waking the container. `cancel` marks the series completed, `pause` holds a pending task, `resume` re-arms a paused one.
+
 ## Manual Container Probes
 
 The container's entry point is `exec bun run /app/src/index.ts`; it talks only to the mounted session DBs, so there is no JSON to pipe in. To probe the image directly:
@@ -258,13 +281,14 @@ docker builder prune -af
 
 ## Clearing a Session
 
-Conversation continuity lives in the container-owned `session_state` table in `outbound.db` (the provider's session/continuation id). The agent's `/clear` clears it. To reset a session from the host, remove the session folder so a fresh one is provisioned on the next message:
+Conversation continuity lives in the container-owned `session_state` table in `outbound.db` (the provider's session/continuation id). The agent's `/clear` clears it. To reset a session from the host, just remove the session folder — it self-heals: the host re-creates the folder and re-initializes both DBs on the next inbound message, so you leave the central `sessions` row in place and don't need to restart the host.
 
 ```bash
 # Inspect first
 ncl sessions get <session-id>
 
-# Remove a single session's folder (host re-provisions both DBs on next message)
+# Remove a single session's folder — the host re-provisions it (both DBs)
+# on the next message. No row deletion or host restart needed.
 rm -rf data/v2-sessions/<group>/<session>/
 ```
 

CLAUDE.md

@@ -105,6 +105,7 @@ ncl help
 | members | list, add, remove | Unprivileged access gate for an agent group |
 | destinations | list, add, remove | Where an agent group can send messages |
 | sessions | list, get | Active sessions (read-only) |
+| tasks | list, get, cancel, pause, resume | Scheduled tasks (cron jobs) in a group's sessions — operator surface to stop a runaway task |
 | user-dms | list | Cold-DM cache (read-only) |
 | dropped-messages | list | Messages from unregistered senders (read-only) |
 | approvals | list, get | Pending approval requests (read-only) |

src/cli/resources/index.ts

@@ -14,3 +14,4 @@ import './user-dms.js';
 import './dropped-messages.js';
 import './approvals.js';
 import './sessions.js';
+import './tasks.js';

src/cli/resources/tasks.test.ts

@@ -0,0 +1,165 @@
+/**
+ * `ncl tasks` operator surface — the only host-side way to inspect and stop
+ * scheduled tasks (cron jobs) without asking the agent that owns them.
+ *
+ * Tasks are `messages_in` rows with `kind='task'` in a session's inbound.db,
+ * so the resource opens each session DB from the host (the legitimate writer),
+ * the same way destinations.ts projects rows. These tests drive dispatch with
+ * `caller: 'host'` — the path a real operator socket connection uses.
+ */
+import Database from 'better-sqlite3';
+import fs from 'fs';
+import { describe, expect, it, beforeEach, afterEach, vi } from 'vitest';
+
+vi.mock('../../container-runner.js', () => ({
+  wakeContainer: vi.fn().mockResolvedValue(undefined),
+  isContainerRunning: vi.fn().mockReturnValue(false),
+  getActiveContainerCount: vi.fn().mockReturnValue(0),
+  killContainer: vi.fn(),
+}));
+
+vi.mock('../../config.js', async () => {
+  const actual = await vi.importActual('../../config.js');
+  return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-cli-tasks' };
+});
+
+const TEST_DIR = '/tmp/nanoclaw-test-cli-tasks';
+
+import { initTestDb, closeDb, runMigrations, createAgentGroup } from '../../db/index.js';
+import { createSession } from '../../db/sessions.js';
+import { insertRecurrence, insertTask } from '../../modules/scheduling/db.js';
+import { initSessionFolder, inboundDbPath, openInboundDb } from '../../session-manager.js';
+import { dispatch } from '../dispatch.js';
+// Side-effect import: registers the `tasks-*` commands.
+import './tasks.js';
+
+const AG = 'ag-tasks';
+const SESSION = 'sess-tasks-1';
+
+function now(): string {
+  return new Date().toISOString();
+}
+
+function readTaskStatus(id: string): string | undefined {
+  const db = new Database(inboundDbPath(AG, SESSION), { readonly: true });
+  const row = db.prepare('SELECT status FROM messages_in WHERE id = ?').get(id) as { status: string } | undefined;
+  db.close();
+  return row?.status;
+}
+
+function seedTask(id: string, prompt: string): void {
+  const db = openInboundDb(AG, SESSION);
+  try {
+    insertTask(db, {
+      id,
+      processAfter: now(),
+      recurrence: null,
+      platformId: null,
+      channelType: null,
+      threadId: null,
+      content: JSON.stringify({ prompt }),
+    });
+  } finally {
+    db.close();
+  }
+}
+
+describe('tasks CLI resource (operator surface)', () => {
+  beforeEach(() => {
+    if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
+    fs.mkdirSync(TEST_DIR, { recursive: true });
+
+    const db = initTestDb();
+    runMigrations(db);
+
+    createAgentGroup({ id: AG, name: 'tasks', folder: 'tasks', agent_provider: null, created_at: now() });
+    createSession({
+      id: SESSION,
+      agent_group_id: AG,
+      messaging_group_id: null,
+      thread_id: null,
+      agent_provider: null,
+      status: 'active',
+      container_status: 'stopped',
+      last_active: null,
+      created_at: now(),
+    });
+    initSessionFolder(AG, SESSION);
+  });
+
+  afterEach(() => {
+    closeDb();
+    if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
+  });
+
+  it('list: returns a pending task from the session inbound.db', async () => {
+    seedTask('task-1', 'water the plants');
+
+    const resp = await dispatch({ id: 'req-list', command: 'tasks-list', args: {} }, { caller: 'host' });
+
+    expect(resp.ok).toBe(true);
+    const rows = (resp as { ok: true; data: unknown }).data as Array<Record<string, unknown>>;
+    expect(rows).toHaveLength(1);
+    expect(rows[0]).toMatchObject({
+      agent_group_id: AG,
+      session_id: SESSION,
+      id: 'task-1',
+      status: 'pending',
+      prompt: 'water the plants',
+    });
+  });
+
+  it('cancel: cancels a task matched by its id', async () => {
+    seedTask('task-1', 'water the plants');
+    expect(readTaskStatus('task-1')).toBe('pending');
+
+    const resp = await dispatch(
+      { id: 'req-cancel', command: 'tasks-cancel', args: { id: 'task-1' } },
+      { caller: 'host' },
+    );
+
+    expect(resp.ok).toBe(true);
+    expect((resp as { ok: true; data: { affected: number } }).data.affected).toBe(1);
+    expect(readTaskStatus('task-1')).toBe('completed');
+  });
+
+  it('cancel: cancels a recurring follow-up matched by series id', async () => {
+    // Original firing plus a live follow-up occurrence that shares the series
+    // id but has a distinct row id — the shape recurrence produces. Cancelling
+    // by the series id must reach the follow-up whose id != the arg.
+    seedTask('series-1', 'daily standup');
+    const db = openInboundDb(AG, SESSION);
+    try {
+      insertRecurrence(
+        db,
+        {
+          id: 'series-1',
+          kind: 'task',
+          content: JSON.stringify({ prompt: 'daily standup' }),
+          recurrence: 'daily',
+          process_after: now(),
+          platform_id: null,
+          channel_type: null,
+          thread_id: null,
+          series_id: 'series-1',
+        },
+        'occ-2',
+        now(),
+      );
+    } finally {
+      db.close();
+    }
+    expect(readTaskStatus('occ-2')).toBe('pending');
+
+    const resp = await dispatch(
+      { id: 'req-cancel-series', command: 'tasks-cancel', args: { id: 'series-1' } },
+      { caller: 'host' },
+    );
+
+    expect(resp.ok).toBe(true);
+    // Both the original row and the follow-up share series_id 'series-1'.
+    expect((resp as { ok: true; data: { affected: number } }).data.affected).toBe(2);
+    expect(readTaskStatus('occ-2')).toBe('completed');
+    expect(readTaskStatus('series-1')).toBe('completed');
+  });
+});

src/cli/resources/tasks.ts

@@ -0,0 +1,235 @@
+import fs from 'fs';
+
+import type Database from 'better-sqlite3';
+
+import { getAllAgentGroups } from '../../db/agent-groups.js';
+import { getSessionsByAgentGroup } from '../../db/sessions.js';
+import { cancelTask, pauseTask, resumeTask } from '../../modules/scheduling/db.js';
+import { inboundDbPath, openInboundDb } from '../../session-manager.js';
+import { registerResource } from '../crud.js';
+import type { CallerContext } from '../frame.js';
+
+/**
+ * `ncl tasks` — the operator surface for scheduled tasks (cron jobs).
+ *
+ * Tasks are `messages_in` rows with `kind='task'` living in each session's
+ * host-owned `inbound.db` — there is no central table, so the auto-CRUD in
+ * crud.ts (which is central-DB-bound) can't back this resource. Instead we
+ * iterate `getSessionsByAgentGroup` and open each session's `inbound.db`
+ * directly from the host process, which is the sole legitimate writer of that
+ * file (see session-manager.ts). This is the same shape destinations.ts uses
+ * to project rows into live sessions.
+ *
+ * Before this existed the only way to stop a runaway recurring task was to ask
+ * the (misbehaving) agent to cancel it via its container-side task tools.
+ */
+
+interface ProjectedTask {
+  agent_group_id: string;
+  session_id: string;
+  /** The stable series handle — one row per series (see list_tasks). */
+  id: string;
+  status: string;
+  process_after: string | null;
+  recurrence: string | null;
+  prompt: string;
+}
+
+/**
+ * Resolve which agent groups the caller may act on. Agent callers are pinned
+ * to their own group (cli_scope=group), mirroring how groups/sessions/
+ * destinations keep a container inside its own agent group. Host (operator)
+ * callers may target one group via --group or every group when it's omitted.
+ */
+function targetAgentGroupIds(group: string | undefined, ctx: CallerContext): string[] {
+  if (ctx.caller === 'agent') return [ctx.agentGroupId];
+  if (group) return [group];
+  return getAllAgentGroups().map((g) => g.id);
+}
+
+/**
+ * Run `cb` against every session's host-owned inbound.db for the resolved
+ * agent groups. Skips sessions whose inbound.db hasn't been created yet
+ * (same guard as write-destinations.ts) so opening one never fabricates an
+ * empty, schema-less DB file.
+ */
+function forEachSessionDb(
+  groupIds: string[],
+  cb: (db: Database.Database, agentGroupId: string, sessionId: string) => void,
+): void {
+  for (const agentGroupId of groupIds) {
+    for (const session of getSessionsByAgentGroup(agentGroupId)) {
+      if (!fs.existsSync(inboundDbPath(agentGroupId, session.id))) continue;
+      const db = openInboundDb(agentGroupId, session.id);
+      try {
+        cb(db, agentGroupId, session.id);
+      } finally {
+        db.close();
+      }
+    }
+  }
+}
+
+function taskPrompt(content: string): string {
+  try {
+    const parsed = JSON.parse(content) as { prompt?: unknown };
+    return typeof parsed.prompt === 'string' ? parsed.prompt : '';
+  } catch {
+    return '';
+  }
+}
+
+/** Count the live rows a control op would touch, so we can report `affected`. */
+function countTargetRows(db: Database.Database, taskId: string, statuses: string[]): number {
+  const placeholders = statuses.map(() => '?').join(', ');
+  const row = db
+    .prepare(
+      `SELECT COUNT(*) AS n FROM messages_in
+        WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status IN (${placeholders})`,
+    )
+    .get(taskId, taskId, ...statuses) as { n: number };
+  return row.n;
+}
+
+/**
+ * Shared body for cancel/pause/resume. `statuses` is the set of live statuses
+ * the underlying db.ts helper actually mutates, used only to report how many
+ * rows were affected. The helper itself matches by id OR series_id, so a
+ * recurring task's live next occurrence is caught, not just the row an agent
+ * happens to remember.
+ */
+function control(
+  args: Record<string, unknown>,
+  ctx: CallerContext,
+  statuses: string[],
+  apply: (db: Database.Database, taskId: string) => void,
+): { taskId: string; affected: number; sessions: string[] } {
+  const taskId = args.id as string | undefined;
+  if (!taskId) throw new Error('--id is required (task id or series id)');
+  const groupIds = targetAgentGroupIds(args.group as string | undefined, ctx);
+
+  let affected = 0;
+  const sessions: string[] = [];
+  forEachSessionDb(groupIds, (db, _agentGroupId, sessionId) => {
+    const n = countTargetRows(db, taskId, statuses);
+    if (n === 0) return;
+    apply(db, taskId);
+    affected += n;
+    sessions.push(sessionId);
+  });
+
+  return { taskId, affected, sessions };
+}
+
+registerResource({
+  name: 'task',
+  plural: 'tasks',
+  // Tasks aren't a central-DB table — they're messages_in rows in per-session
+  // inbound.db files. `table` is unused because no generic CRUD op is enabled.
+  table: 'messages_in',
+  description:
+    "Scheduled task (cron job) — a messages_in row with kind=task in a session inbound.db. Operator surface to inspect and stop tasks without going through the agent. list/get show one row per series; cancel/pause/resume match by id OR series id so a recurring task's live next occurrence is caught.",
+  idColumn: 'id',
+  scopeField: 'agent_group_id',
+  columns: [
+    { name: 'agent_group_id', type: 'string', description: 'Agent group whose session holds the task.' },
+    { name: 'session_id', type: 'string', description: 'Session whose inbound.db holds the task.' },
+    { name: 'id', type: 'string', description: 'Series id — the stable handle for the task.' },
+    { name: 'status', type: 'string', description: '"pending" or "paused".' },
+    { name: 'process_after', type: 'string', description: 'When the next occurrence is due (UTC).' },
+    { name: 'recurrence', type: 'string', description: 'Recurrence rule, or null for a one-shot.' },
+    { name: 'prompt', type: 'string', description: 'The task prompt the agent runs when it fires.' },
+  ],
+  operations: {},
+  customOperations: {
+    list: {
+      access: 'open',
+      description:
+        'List pending and paused scheduled tasks (one row per series). Use --group to scope to a single agent group; defaults to all groups.',
+      handler: async (args, ctx) => {
+        const groupIds = targetAgentGroupIds(args.group as string | undefined, ctx);
+        const tasks: ProjectedTask[] = [];
+        forEachSessionDb(groupIds, (db, agentGroupId, sessionId) => {
+          const rows = db
+            .prepare(
+              `SELECT series_id AS id, status, process_after, recurrence, content, MAX(seq) AS _seq
+                 FROM messages_in
+                WHERE kind = 'task' AND status IN ('pending', 'paused')
+                GROUP BY series_id
+                ORDER BY process_after ASC`,
+            )
+            .all() as Array<{
+            id: string;
+            status: string;
+            process_after: string | null;
+            recurrence: string | null;
+            content: string;
+          }>;
+          for (const r of rows) {
+            tasks.push({
+              agent_group_id: agentGroupId,
+              session_id: sessionId,
+              id: r.id,
+              status: r.status,
+              process_after: r.process_after,
+              recurrence: r.recurrence,
+              prompt: taskPrompt(r.content),
+            });
+          }
+        });
+        return tasks;
+      },
+    },
+    get: {
+      access: 'open',
+      description: 'Get a scheduled task by id or series id. Use --id <task-id>, optional --group.',
+      handler: async (args, ctx) => {
+        const taskId = args.id as string | undefined;
+        if (!taskId) throw new Error('--id is required (task id or series id)');
+        const groupIds = targetAgentGroupIds(args.group as string | undefined, ctx);
+        let found: ProjectedTask | undefined;
+        forEachSessionDb(groupIds, (db, agentGroupId, sessionId) => {
+          if (found) return;
+          const r = db
+            .prepare(
+              `SELECT series_id AS id, status, process_after, recurrence, content, MAX(seq) AS _seq
+                 FROM messages_in
+                WHERE kind = 'task' AND (id = ? OR series_id = ?) AND status IN ('pending', 'paused')
+                GROUP BY series_id`,
+            )
+            .get(taskId, taskId) as
+            | { id: string; status: string; process_after: string | null; recurrence: string | null; content: string }
+            | undefined;
+          if (r) {
+            found = {
+              agent_group_id: agentGroupId,
+              session_id: sessionId,
+              id: r.id,
+              status: r.status,
+              process_after: r.process_after,
+              recurrence: r.recurrence,
+              prompt: taskPrompt(r.content),
+            };
+          }
+        });
+        if (!found) throw new Error(`task not found: ${taskId}`);
+        return found;
+      },
+    },
+    cancel: {
+      access: 'approval',
+      description: 'Cancel a scheduled task (matches id or series id). Use --id <task-id>, optional --group.',
+      handler: async (args, ctx) => control(args, ctx, ['pending', 'paused'], cancelTask),
+    },
+    pause: {
+      access: 'approval',
+      description: 'Pause a scheduled task so it stops firing until resumed. Use --id <task-id>, optional --group.',
+      handler: async (args, ctx) => control(args, ctx, ['pending'], pauseTask),
+    },
+    resume: {
+      access: 'approval',
+      description: 'Resume a paused scheduled task. Use --id <task-id>, optional --group.',
+      handler: async (args, ctx) => control(args, ctx, ['paused'], resumeTask),
+    },
+  },
+});