Compare commits

...

2 Commits

Author SHA1 Message Date
omri-maya 5459925c65 Apply suggestions from code review
Co-authored-by: omri-maya <omri@nanoco.ai>
2026-07-09 11:23:08 +03:00
Omri Maya b0c76ce4c5 feat(tasks): ncl tasks control plane — isolated per-series sessions, strict verb contract, script gate
Scheduled tasks move from MCP tools to a first-class ncl resource:
list / get / create / update / cancel / pause / resume / run / append-log,
with args declared on every verb (strict validation, fix-carrying errors)
and deep per-verb help with executable examples.

A series is CronJob-like: the live (pending/paused) row is the next
occurrence, completed rows are its run history. tasks list reads as a
compact run-history table, server-rendered so the container agent prints
the same aligned table. Short named ids (<slug>-<hex>) are
filesystem/thread-safe and copy-pasteable. Agents keep a per-series work
journal: the run log at tasks/<series>.md (append-log + auto-logged
final text).

Each series runs in its own isolated system session
(system:tasks:<series>); the live-task cap is dropped — isolation
replaces throttling. Spent task sessions are GC'd by the sweep.

--script on a task runs a bash gate BEFORE the agent wakes. It prints a
JSON verdict as its last stdout line: {"wakeAgent": bool, "data": {...}}.
wakeAgent:false handles the occurrence without waking the agent; data is
threaded into the fire's prompt, so a gate can fetch/inspect and hand the
agent exactly what changed. Failing scripts back off exponentially
(2·2^(n-1) min, cap 60) and auto-pause the series after 8 consecutive
failures with a host-written run-log note; manual runs are excluded from
the streak.

BREAKING: the schedule_task / list_tasks / update_task / cancel_task /
pause_task / resume_task MCP tools are removed. Agents schedule via
ncl tasks (available under cli_scope=group). Existing task rows keep
firing — storage (messages_in kind='task') is unchanged; only the
management surface moved. Migration: docs/ncl-tasks-migration.md.
Review round (gavrielc):
- Recurrence frequency guard: create/update refuse a recurrence more
  frequent than 4 fires/day with a warning steering to gate scripts;
  --dangerously-override-recurrence-limit bypasses after explicit user
  confirmation. Counted over the next 24h in the instance TZ. Guarded on
  update too, so create-slow-then-update cannot sneak past.
- ncl tasks help scripts: resource help topics (new helpTopics seam on
  registerResource) carry the full gate-script guide — contract, examples,
  state, failure/backoff semantics, testing directions.
- Agent-facing scheduling instructions slimmed: no recurring-frequency
  prose in the fragment; one pointer to the help topic.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-08 17:04:45 +03:00
46 changed files with 2374 additions and 706 deletions
+2
View File
@@ -4,6 +4,8 @@ All notable changes to NanoClaw will be documented in this file.
## [Unreleased]
- **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. Also fixed: an explicitly-addressed `<message to>` in a task fire's final text now delivers as a deliberate send (previously suppressed as a turn-reply echo → zero delivery when the agent skipped the MCP tool); identical echoes of an MCP send are dropped in the runner, where the duplication originates.
- [BREAKING] **Scheduled tasks moved from MCP tools to `ncl tasks`.** The six scheduling MCP tools are no longer exposed to agent containers; agents and operators manage tasks with `ncl tasks list/get/create/update/cancel/pause/resume/delete`. New tasks run from a per-agent-group system session rather than waking the chat session that created them, and task writes are not approval-gated inside the owning group. **Migration:** [docs/ncl-tasks-migration.md](docs/ncl-tasks-migration.md).
- **Optional per-container resource caps.** `CONTAINER_CPU_LIMIT` and `CONTAINER_MEMORY_LIMIT` pass through to `docker run` as `--cpus` / `--memory` (`container-runner.ts`). Both empty by default — no flag added, spawn args byte-identical to today — so existing installs are unaffected. Set them to cap an agent container's CPU/memory so one agent can't monopolize the host (e.g. `CONTAINER_CPU_LIMIT=2`, `CONTAINER_MEMORY_LIMIT=8g`). Swap is intentionally not managed here: `--memory` is a hard cap on a swapless host.
- [BREAKING] **Chat SDK pinned to `4.29.0` (was `4.26.0` via `^4.24.0`).** `chat` and the `@chat-adapter/*` channel adapters are version-locked — the adapter's `ChatInstance` must match the bridge's, so a mismatched pair fails to typecheck at `createChatSdkBridge(...)`. `chat` is therefore pinned exactly, and the channel-adapter install pins move with it — the `/add-<channel>` SKILL.md steps and `setup/*.sh` scripts on `main`, plus the adapter code on the `channels` branch. Core installs with no channel (only `cli`) are unaffected. **Migration:** if any channel is installed (Slack, Discord, Telegram, Teams, …), re-run its `/add-<channel>` skill to pull the matching `4.29.0` adapter.
- **Budget/billing-exhausted LLM turns now reach the user instead of being silently dropped.** When a turn ends in a non-retryable provider error (e.g. an Anthropic `403 billing_error`) with no `<message>` wrapping, the agent-runner delivers the provider's notice to the originating channel and stops re-nudging the failing gateway. `providers/claude.ts` now surfaces the SDK's `is_error` flag (and the error subtype's `errors[]` text); `poll-loop.ts` delivers that text and skips the re-wrap retry. Fixes the case where a spend-limit notice produced silence plus a turn-after-turn retry loop.
+2 -1
View File
@@ -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, create, update, cancel, pause, resume, delete, run, append-log | Scheduled tasks for an agent group |
| 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) |
@@ -137,7 +138,7 @@ Per-agent-group container runtime config (provider, model, packages, MCP servers
| Value | Behavior |
|-------|----------|
| `disabled` | Agent never learns about ncl (instructions excluded from CLAUDE.md). Host dispatch rejects any `cli_request`. |
| `group` (default) | Agent can access `groups`, `sessions`, `destinations`, `members` only, scoped to its own agent group. `--id` and group args are auto-filled. Cross-group access rejected. `cli_scope` changes blocked. |
| `group` (default) | Agent can access `groups`, `sessions`, `destinations`, `members`, `tasks` only, scoped to its own agent group. `--id` and group args are auto-filled. Cross-group access rejected. `cli_scope` changes blocked. |
| `global` | Unrestricted. Set automatically for owner agent groups via `init-first-agent`. |
Key files: `src/db/container-configs.ts`, `src/container-config.ts`, `src/cli/dispatch.ts` (scope enforcement), `src/claude-md-compose.ts` (instructions exclusion).
@@ -120,6 +120,24 @@ export function markCompleted(ids: string[]): void {
})();
}
/**
* Ack task messages whose pre-task script gated the run. The reason decides
* the ack: `gated` (wakeAgent=false) is the monitor working as designed → a
* plain `completed`; `error` (broken script) → `script-skip:error`, which the
* host's ack sync records as a FAILED run so recurrence can read the trailing
* failed streak off the occurrence rows and back the series off.
*/
export function markScriptSkipped(skips: Array<{ id: string; reason: string }>): void {
if (skips.length === 0) return;
const db = getOutboundDb();
const stmt = db.prepare(
"INSERT OR REPLACE INTO processing_ack (message_id, status, status_changed) VALUES (?, ?, datetime('now'))",
);
db.transaction(() => {
for (const s of skips) stmt.run(s.id, s.reason === 'error' ? 'script-skip:error' : 'completed');
})();
}
/** Mark a single message as failed — writes to processing_ack in outbound.db. */
export function markFailed(id: string): void {
getOutboundDb()
@@ -141,3 +141,22 @@ export function getUndeliveredMessages(): MessageOutRow[] {
)
.all() as MessageOutRow[];
}
/**
* True if a deliberate send with this exact destination + text already exists
* (an MCP send_message row from the current turn). Used by the task-fire
* final-text dispatcher to drop the turn-final <message> echo of a send the
* agent already made — the dedup happens where the duplication originates.
*/
export function hasIdenticalSend(platformId: string, channelType: string, text: string): boolean {
const row = getOutboundDb()
.prepare(
`SELECT 1 FROM messages_out
WHERE platform_id = $platform_id AND channel_type = $channel_type
AND (in_reply_to IS NULL OR in_reply_to = '')
AND json_extract(content, '$.text') = $text
LIMIT 1`,
)
.get({ $platform_id: platformId, $channel_type: channelType, $text: text });
return row != null;
}
+9 -4
View File
@@ -105,13 +105,11 @@ function buildDestinationsSection(): string {
const lines = ['## Sending messages', ''];
if (all.length === 1) {
const d = all[0];
const label = d.displayName && d.displayName !== d.name ? ` (${d.displayName})` : '';
lines.push(`Your destination is \`${d.name}\`${label}.`);
lines.push(`Your destination is \`${d.name}\`${destinationLabel(d)}.`);
} else {
lines.push('You can send messages to the following destinations:', '');
for (const d of all) {
const label = d.displayName && d.displayName !== d.name ? ` (${d.displayName})` : '';
lines.push(`- \`${d.name}\`${label}`);
lines.push(`- \`${d.name}\`${destinationLabel(d)}`);
}
}
lines.push('');
@@ -128,3 +126,10 @@ function buildDestinationsSection(): string {
);
return lines.join('\n');
}
function destinationLabel(d: DestinationEntry): string {
const parts: string[] = [];
if (d.channelType) parts.push(d.channelType);
if (d.displayName && d.displayName !== d.name) parts.push(d.displayName);
return parts.length > 0 ? ` (${parts.join(' · ')})` : '';
}
+6
View File
@@ -98,6 +98,11 @@ export interface RoutingContext {
channelType: string | null;
threadId: string | null;
inReplyTo: string | null;
/** Batch is a task fire — explicit `<message to>` sends must NOT inherit
* inReplyTo (the series id), or the host's task-fire suppression drops
* them as turn-final echoes: zero delivery. Deliberate sends are
* in_reply_to-null, same as the out-of-process MCP send_message path. */
taskFire: boolean;
}
/**
@@ -111,6 +116,7 @@ export function extractRouting(messages: MessageInRow[]): RoutingContext {
channelType: first?.channel_type ?? null,
threadId: first?.thread_id ?? null,
inReplyTo: first?.id ?? null,
taskFire: messages.length > 0 && messages.every((m) => m.kind === 'task'),
};
}
@@ -18,12 +18,13 @@ Your CLI access may be scoped. Run `ncl help` to see which resources are availab
Run `ncl help` for the full list. Common resources:
| Resource | Verbs | What it is |
|----------|-------|------------|
| groups | list, get, create, update, delete, restart, config get/update, config add-mcp-server/remove-mcp-server, config add-package/remove-package | Agent groups (workspace, personality, container config) |
| sessions | list, get | Active sessions (read-only) |
| destinations | list, add, remove | Where an agent group can send messages |
| members | list, add, remove | Unprivileged access gate for an agent group |
| Resource | Verbs | What it is |
| ------------ | ----------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------- |
| groups | list, get, create, update, delete, restart, config get/update, config add-mcp-server/remove-mcp-server, config add-package/remove-package | Agent groups (workspace, personality, container config) |
| sessions | list, get | Active sessions (read-only) |
| destinations | list, add, remove | Where an agent group can send messages |
| members | list, add, remove | Unprivileged access gate for an agent group |
| tasks | list, get, create, update, cancel, pause, resume, delete, append-log | Scheduled tasks for your agent group |
Additional resources (available under `global` scope only): messaging-groups, wirings, users, roles, user-dms, dropped-messages, approvals.
@@ -33,11 +34,12 @@ Additional resources (available under `global` scope only): messaging-groups, wi
- **Restarting your container**`ncl groups restart` (with optional `--rebuild` and `--message`).
- **Checking who's in your group**`ncl members list`.
- **Seeing your destinations**`ncl destinations list`.
- **Scheduling work**`ncl tasks create`, then `ncl tasks list/get/update/cancel/pause/resume/delete`; `ncl tasks run <id>` fires one extra run now (testing) without changing the schedule. At the end of each task run, `ncl tasks append-log --msg "…"` to record what happened (host-timestamped, not a message).
- **Answering questions about the system** — query `ncl` rather than guessing.
### Access rules
Read commands (list, get) are open. Write commands (create, update, delete, restart, config update, add, remove) require admin approval — the request is held until an admin approves it.
Read commands (list, get) are open. Most write commands (create, update, delete, restart, config update, add, remove) require admin approval — the request is held until an admin approves it. `ncl tasks` is the exception: an agent can manage its own group tasks without approval.
### Approval flow
@@ -61,6 +63,13 @@ ncl groups config get
ncl sessions list
ncl destinations list
ncl members list
ncl tasks list
# Always pass a short descriptive --name so the task id is readable (e.g. daily-briefing-a25c, not a long uuid).
# For a recurring task, --recurrence alone sets the schedule (first run derived from it); add --process-after only for one-shots.
ncl tasks create --name "daily briefing" --prompt "Send the daily briefing" --recurrence "0 9 * * *"
# At the END of every task run, record one line of history. The host stamps the UTC time — you supply only the summary.
# This is a LOG ENTRY, not a message: it sends nothing to anyone. Inside a task fire --id is auto-derived from your session.
ncl tasks append-log --msg "posted the daily digest to slack; one feed returned 403, skipped"
# Write commands (approval required)
ncl groups restart
@@ -6,7 +6,6 @@
* at module scope, and append the import here. No central list.
*/
import './core.js';
import './scheduling.js';
import './interactive.js';
import './agents.js';
import './self-mod.js';
@@ -1,40 +1,25 @@
## Task scheduling (`schedule_task`)
## Task scheduling (`ncl tasks`)
For any recurring task, use `schedule_task`. This is the scheduling path — tasks persist across sessions and restarts, and support the pre-task `script` hook described below.
Use `ncl tasks` for one-shot and recurring tasks. A task runs in this agent group's system session, not in the current chat session, so when it fires you must choose a destination explicitly with `<message to="name">...</message>` or `send_message({ to: "name", ... })`.
To inspect or change existing tasks, use `list_tasks` (returns one row per series with the stable id) and `update_task` / `cancel_task` / `pause_task` / `resume_task`. Prefer `update_task` over cancel + reschedule.
Pass `--name "<short label>"` on create to get a readable task id (e.g. `--name "sales briefing"` `sales-briefing-a25c`); without it ids are `t-<hex>`.
Frequent recurring scheduled tasks — more than a few times a day — consume API credits and can risk account restrictions. You can add a `script` that runs first, and you will only be called when the check passes.
### How it works
1. Provide a bash `script` alongside the `prompt` when scheduling
2. When the task fires, the script runs first
3. Script returns: `{ "wakeAgent": true/false, "data": {...} }`
4. If `wakeAgent: false` — nothing happens, task waits for next run
5. If `wakeAgent: true` — claude receives the script's data + prompt and handles
### Always test your script first
Before scheduling, run the script directly to verify it works:
Common commands:
```bash
bash -c 'node --input-type=module -e "
const r = await fetch(\"https://api.github.com/repos/owner/repo/pulls?state=open\");
const prs = await r.json();
console.log(JSON.stringify({ wakeAgent: prs.length > 0, data: prs.slice(0, 5) }));
"'
ncl tasks create --name "ping" --prompt "Remind me to call Dana" --process-after "tomorrow 18:00"
ncl tasks list
ncl tasks get ping-a25c # includes run count, failures, and recent run-log lines
ncl tasks run ping-a25c # fire once now without changing the schedule (testing)
ncl tasks update ping-a25c --prompt "New instructions"
ncl tasks pause ping-a25c
ncl tasks resume ping-a25c
ncl tasks cancel ping-a25c # or --all as a kill switch
ncl tasks delete ping-a25c
```
### When NOT to use scripts
Use good judgement on whether it's appropriate to check in with the user about the task prompt before task creation, and if so, whether to share verbatim or a description of it.
If a task requires your judgment every time (daily briefings, reminders, reports), skip the script — just use a regular prompt. Do not attempt to do things like sentiment analysis or advanced nlp in scripts.
`--process-after` accepts UTC timestamps or naive local timestamps interpreted in the instance timezone (shown in the `<context timezone="..."/>` header).
### Frequent task guidance
If a user wants a task to run more than a few times a day and a script can't be used:
- Explain that each time the task fires it uses API credits and risks rate limits
- Suggest adjusting the task requirements in a way that will allow you to use a script
- If the user needs an LLM to evaluate data, suggest using an API key with direct Anthropic API calls inside the script
- Help the user find the minimum viable frequency
Run `ncl tasks create --help` for schedules, options, and pre-task gate scripts (checks that run before you wake).
@@ -1,302 +0,0 @@
/**
* Scheduling MCP tools: schedule_task, list_tasks, cancel_task, pause_task, resume_task.
*
* With the two-DB split, the container cannot write to inbound.db (host-owned).
* Scheduling operations are sent as system actions via messages_out the host
* reads them during delivery and applies the changes to inbound.db.
*/
import { getInboundDb } from '../db/connection.js';
import { writeMessageOut } from '../db/messages-out.js';
import { getSessionRouting } from '../db/session-routing.js';
import { TIMEZONE, parseZonedToUtc } from '../timezone.js';
import { registerTools } from './server.js';
import type { McpToolDefinition } from './types.js';
function log(msg: string): void {
console.error(`[mcp-tools] ${msg}`);
}
function generateId(): string {
return `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
}
function routing() {
return getSessionRouting();
}
function ok(text: string) {
return { content: [{ type: 'text' as const, text }] };
}
function err(text: string) {
return { content: [{ type: 'text' as const, text: `Error: ${text}` }], isError: true };
}
export const scheduleTask: McpToolDefinition = {
tool: {
name: 'schedule_task',
description:
`Schedule a one-shot or recurring task. The user's timezone is declared in the <context timezone="..."/> header of your prompt — interpret the user's "9pm" etc. in that zone. Cron expressions are interpreted in the user's timezone too.`,
inputSchema: {
type: 'object' as const,
properties: {
prompt: { type: 'string', description: 'Task instructions/prompt' },
processAfter: {
type: 'string',
description:
`ISO 8601 timestamp for the first run. Accepts either UTC (ending in "Z" or "+00:00") or a naive local timestamp (no offset) which is interpreted in the user's timezone (e.g. "2026-01-15T21:00:00" = 9pm user-local). Prefer naive local.`,
},
recurrence: {
type: 'string',
description:
'Cron expression for recurring tasks (e.g., "0 9 * * 1-5" = weekdays at 9am user-local). Evaluated in the user\'s timezone.',
},
script: { type: 'string', description: 'Optional pre-agent script to run before processing' },
},
required: ['prompt', 'processAfter'],
},
},
async handler(args) {
const prompt = args.prompt as string;
const processAfterIn = args.processAfter as string;
if (!prompt || !processAfterIn) return err('prompt and processAfter are required');
let processAfter: string;
try {
const d = parseZonedToUtc(processAfterIn, TIMEZONE);
if (Number.isNaN(d.getTime())) return err(`invalid processAfter: ${processAfterIn}`);
processAfter = d.toISOString();
} catch {
return err(`invalid processAfter: ${processAfterIn}`);
}
const id = generateId();
const r = routing();
const recurrence = (args.recurrence as string) || null;
const script = (args.script as string) || null;
// Write as a system action — host will insert into inbound.db
writeMessageOut({
id,
kind: 'system',
platform_id: r.platform_id,
channel_type: r.channel_type,
thread_id: r.thread_id,
content: JSON.stringify({
action: 'schedule_task',
taskId: id,
prompt,
script,
processAfter,
recurrence,
platformId: r.platform_id,
channelType: r.channel_type,
threadId: r.thread_id,
}),
});
log(`schedule_task: ${id} at ${processAfter}${recurrence ? ` (recurring: ${recurrence})` : ''}`);
return ok(`Task scheduled (id: ${id}, runs at: ${processAfter}${recurrence ? `, recurrence: ${recurrence}` : ''})`);
},
};
export const listTasks: McpToolDefinition = {
tool: {
name: 'list_tasks',
description:
'List scheduled tasks. Returns one row per series — the live (pending or paused) occurrence. The id shown is the series id, which is what update_task / cancel_task / pause_task / resume_task expect.',
inputSchema: {
type: 'object' as const,
properties: {
status: { type: 'string', description: 'Filter by status: pending or paused (default: both)' },
},
},
},
async handler(args) {
const status = args.status as string | undefined;
const db = getInboundDb();
// One row per series — the live (pending or paused) occurrence. Recurring
// tasks accumulate one completed row per firing plus one live follow-up;
// exposing the whole pile to the agent is noisy and confuses task identity
// ("which id do I cancel?"). The series_id is the stable handle.
//
// SQLite quirk: when MAX(seq) appears in the SELECT list of a GROUP BY
// query, the bare columns take values from the row that contains that max
// — that's how we pick "the latest live row per series" in one pass.
let rows;
if (status) {
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 = ?
GROUP BY series_id
ORDER BY process_after ASC`,
)
.all(status);
} else {
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();
}
if ((rows as unknown[]).length === 0) return ok('No tasks found.');
const lines = (rows as Array<{ id: string; status: string; process_after: string | null; recurrence: string | null; content: string }>).map((r) => {
const content = JSON.parse(r.content);
const prompt = (content.prompt as string || '').slice(0, 80);
return `- ${r.id} [${r.status}] at=${r.process_after || 'now'} ${r.recurrence ? `recur=${r.recurrence} ` : ''}${prompt}`;
});
return ok(lines.join('\n'));
},
};
export const cancelTask: McpToolDefinition = {
tool: {
name: 'cancel_task',
description: 'Cancel a scheduled task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to cancel' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
// Write as a system action — host will update inbound.db
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'cancel_task', taskId }),
});
log(`cancel_task: ${taskId}`);
return ok(`Task cancellation requested: ${taskId}`);
},
};
export const pauseTask: McpToolDefinition = {
tool: {
name: 'pause_task',
description: 'Pause a scheduled task. It will not run until resumed.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to pause' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'pause_task', taskId }),
});
log(`pause_task: ${taskId}`);
return ok(`Task pause requested: ${taskId}`);
},
};
export const resumeTask: McpToolDefinition = {
tool: {
name: 'resume_task',
description: 'Resume a paused task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to resume' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'resume_task', taskId }),
});
log(`resume_task: ${taskId}`);
return ok(`Task resume requested: ${taskId}`);
},
};
export const updateTask: McpToolDefinition = {
tool: {
name: 'update_task',
description:
'Update a scheduled task. Pass the series id from list_tasks. Any field omitted is left unchanged. Use this instead of cancel + reschedule when adjusting an existing task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Series id of the task to update (as shown by list_tasks)' },
prompt: { type: 'string', description: 'New task prompt (optional)' },
recurrence: {
type: 'string',
description: 'New cron expression (optional). Pass empty string to clear and make the task one-shot.',
},
processAfter: {
type: 'string',
description:
`New ISO 8601 timestamp for the next run (optional). Accepts either UTC (ending in "Z" / "+00:00") or a naive local timestamp interpreted in the user's timezone.`,
},
script: {
type: 'string',
description: 'New pre-agent script (optional). Pass empty string to clear.',
},
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
const update: Record<string, unknown> = { taskId };
if (typeof args.prompt === 'string') update.prompt = args.prompt;
if (typeof args.processAfter === 'string') {
try {
const d = parseZonedToUtc(args.processAfter, TIMEZONE);
if (Number.isNaN(d.getTime())) return err(`invalid processAfter: ${args.processAfter}`);
update.processAfter = d.toISOString();
} catch {
return err(`invalid processAfter: ${args.processAfter}`);
}
}
// Empty string clears recurrence/script; undefined leaves them as-is.
if (typeof args.recurrence === 'string') update.recurrence = args.recurrence === '' ? null : args.recurrence;
if (typeof args.script === 'string') update.script = args.script === '' ? null : args.script;
if (Object.keys(update).length === 1) return err('at least one field to update is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'update_task', ...update }),
});
log(`update_task: ${taskId}`);
return ok(`Task update requested: ${taskId}`);
},
};
registerTools([scheduleTask, listTasks, updateTask, cancelTask, pauseTask, resumeTask]);
+19 -10
View File
@@ -1,6 +1,6 @@
import { findByName, getAllDestinations, type DestinationEntry } from './destinations.js';
import { getPendingMessages, markProcessing, markCompleted, type MessageInRow } from './db/messages-in.js';
import { writeMessageOut } from './db/messages-out.js';
import { getPendingMessages, markProcessing, markCompleted, markScriptSkipped, type MessageInRow } from './db/messages-in.js';
import { hasIdenticalSend, writeMessageOut } from './db/messages-out.js';
import { getInboundDb, touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js';
import {
clearContinuation,
@@ -207,15 +207,15 @@ export async function runPollLoop(config: PollLoopConfig): Promise<void> {
// Without the scheduling module, the marker block is empty, `keep`
// falls back to `normalMessages`, and no gating happens.
let keep: MessageInRow[] = normalMessages;
let skipped: string[] = [];
let skipped: Array<{ id: string; reason: string }> = [];
// MODULE-HOOK:scheduling-pre-task:start
const { applyPreTaskScripts } = await import('./scheduling/task-script.js');
const preTask = await applyPreTaskScripts(normalMessages);
keep = preTask.keep;
skipped = preTask.skipped;
if (skipped.length > 0) {
markCompleted(skipped);
log(`Pre-task script skipped ${skipped.length} task(s): ${skipped.join(', ')}`);
markScriptSkipped(skipped);
log(`Pre-task script skipped ${skipped.length} task(s): ${skipped.map((s) => s.id).join(', ')}`);
}
// MODULE-HOOK:scheduling-pre-task:end
@@ -238,7 +238,7 @@ export async function runPollLoop(config: PollLoopConfig): Promise<void> {
});
// Process the query while concurrently polling for new messages
const skippedSet = new Set(skipped);
const skippedSet = new Set(skipped.map((s) => s.id));
const processingIds = ids.filter((id) => !commandIds.includes(id) && !skippedSet.has(id));
// Publish the batch's in_reply_to so MCP tools (send_message, send_file)
// can stamp it on outbound rows — needed for a2a return-path routing.
@@ -401,15 +401,15 @@ export async function processQuery(
// its script gate and always wakes the agent, defeating the gate.
// Mirrors the initial-batch hook above.
let keep = newMessages;
let skipped: string[] = [];
let skipped: Array<{ id: string; reason: string }> = [];
// MODULE-HOOK:scheduling-pre-task-followup:start
const { applyPreTaskScripts } = await import('./scheduling/task-script.js');
const preTask = await applyPreTaskScripts(newMessages);
keep = preTask.keep;
skipped = preTask.skipped;
if (skipped.length > 0) {
markCompleted(skipped);
log(`Pre-task script skipped ${skipped.length} follow-up task(s): ${skipped.join(', ')}`);
markScriptSkipped(skipped);
log(`Pre-task script skipped ${skipped.length} follow-up task(s): ${skipped.map((s) => s.id).join(', ')}`);
}
// MODULE-HOOK:scheduling-pre-task-followup:end
@@ -650,6 +650,15 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb
function sendToDestination(dest: DestinationEntry, body: string, routing: RoutingContext): void {
const platformId = dest.type === 'channel' ? dest.platformId! : dest.agentGroupId!;
const channelType = dest.type === 'channel' ? dest.channelType! : 'agent';
// Task fires: an explicitly-addressed final-text block is either the echo of
// an MCP send the agent already made this turn (drop it HERE, where the
// duplication originates) or the agent's only deliberate send (write it
// in_reply_to-null like the MCP path, or the host's task-fire suppression
// would discard it — zero delivery).
if (routing.taskFire && hasIdenticalSend(platformId, channelType, body)) {
log(`Dropping turn-final echo of an already-sent task message to ${dest.name}`);
return;
}
// Resolve thread_id per-destination from the most recent inbound message
// that came from this same channel+platform. In agent-shared sessions,
// different destinations have different thread contexts — using a single
@@ -657,7 +666,7 @@ function sendToDestination(dest: DestinationEntry, body: string, routing: Routin
const destRouting = resolveDestinationThread(channelType, platformId);
writeMessageOut({
id: generateId(),
in_reply_to: destRouting?.inReplyTo ?? routing.inReplyTo,
in_reply_to: destRouting?.inReplyTo ?? (routing.taskFire ? null : routing.inReplyTo),
kind: 'chat',
platform_id: platformId,
channel_type: channelType,
@@ -17,7 +17,7 @@ function log(msg: string): void {
// Code's interactive UI and would hang here).
//
// - CronCreate / CronDelete / CronList / ScheduleWakeup: we have durable
// scheduling via mcp__nanoclaw__schedule_task.
// scheduling via `ncl tasks`.
// - AskUserQuestion: SDK returns a placeholder instead of blocking on a
// real answer — we have mcp__nanoclaw__ask_user_question that persists
// the question and blocks on the real reply.
@@ -0,0 +1,72 @@
/**
* Container leg of the script-failure backoff chain, tested at unit level so
* the e2e suite doesn't need a live multi-sweep scenario for it:
*
* script error applyPreTaskScripts skips with reason 'error'
* markScriptSkipped acks `script-skip:error` in outbound.db
* (gated plain 'completed': the monitor working as designed).
*
* The host leg (ack FAILED run streak backoff) is pinned in
* src/db/session-db.test.ts and src/modules/scheduling/recurrence.test.ts
* both sides pin the literal 'script-skip:error'; if either renames it, its
* own test goes red.
*/
import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from '../db/connection.js';
import { getPendingMessages, markScriptSkipped } from '../db/messages-in.js';
import { applyPreTaskScripts } from './task-script.js';
beforeEach(() => {
initTestSessionDb();
});
afterEach(() => {
closeSessionDb();
});
function insertTask(id: string, script: string) {
getInboundDb()
.prepare(
`INSERT INTO messages_in (id, kind, timestamp, status, trigger, content)
VALUES (?, 'task', datetime('now'), 'pending', 1, ?)`,
)
.run(id, JSON.stringify({ prompt: 'monitor', script }));
}
const ackStatus = (id: string): string | undefined =>
(getOutboundDb().prepare('SELECT status FROM processing_ack WHERE message_id = ?').get(id) as { status: string } | undefined)
?.status;
describe('script-skip ack chain (container leg)', () => {
it('an erroring script skips with reason "error" and acks script-skip:error', async () => {
insertTask('t-err', 'echo boom >&2; exit 1');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(keep).toHaveLength(0);
expect(skipped).toEqual([{ id: 't-err', reason: 'error' }]);
markScriptSkipped(skipped);
expect(ackStatus('t-err')).toBe('script-skip:error');
});
it('a deliberate wakeAgent=false gate acks plain completed — never backs off', async () => {
insertTask('t-gated', 'echo \'{"wakeAgent": false}\'');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(keep).toHaveLength(0);
expect(skipped).toEqual([{ id: 't-gated', reason: 'gated' }]);
markScriptSkipped(skipped);
expect(ackStatus('t-gated')).toBe('completed');
});
it('wakeAgent=true keeps the task and enriches the prompt with script data', async () => {
insertTask('t-wake', 'echo \'{"wakeAgent": true, "data": {"alerts": 2}}\'');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(skipped).toHaveLength(0);
expect(keep).toHaveLength(1);
expect(JSON.parse(keep[0].content).scriptOutput).toEqual({ alerts: 2 });
});
});
@@ -64,21 +64,26 @@ export async function runScript(script: string, taskId: string): Promise<ScriptR
});
}
/** Why a script gated its task: deliberate wakeAgent=false vs a broken script. */
export type ScriptSkipReason = 'gated' | 'error';
export interface TaskScriptOutcome {
keep: MessageInRow[];
skipped: string[];
skipped: Array<{ id: string; reason: ScriptSkipReason }>;
}
/**
* Run pre-task scripts for any task messages that carry one, serially.
* - Errors / missing output / wakeAgent=false task id added to `skipped`.
* - Errors / missing output / wakeAgent=false task id added to `skipped`,
* with the reason. The caller acks these as script-skips (not plain
* completions) so the host can count consecutive failures and back off.
* - wakeAgent=true content JSON is mutated to carry `scriptOutput`, so the
* formatter renders it into the prompt.
* Non-task messages and tasks without scripts pass through unchanged.
*/
export async function applyPreTaskScripts(messages: MessageInRow[]): Promise<TaskScriptOutcome> {
const keep: MessageInRow[] = [];
const skipped: string[] = [];
const skipped: Array<{ id: string; reason: ScriptSkipReason }> = [];
for (const msg of messages) {
if (msg.kind !== 'task') {
@@ -106,9 +111,9 @@ export async function applyPreTaskScripts(messages: MessageInRow[]): Promise<Tas
touchHeartbeat();
if (!result || !result.wakeAgent) {
const reason = result ? 'wakeAgent=false' : 'script error/no output';
log(`task ${msg.id} skipped: ${reason}`);
skipped.push(msg.id);
const reason: ScriptSkipReason = result ? 'gated' : 'error';
log(`task ${msg.id} skipped: ${reason === 'gated' ? 'wakeAgent=false' : 'script error/no output'}`);
skipped.push({ id: msg.id, reason });
continue;
}
+3 -3
View File
@@ -134,11 +134,11 @@ A personal AI assistant accessible via messaging, with minimal custom code.
### Scheduler
- Built-in scheduler runs on the host, spawns containers for task execution
- Custom `nanoclaw` MCP server (inside container) provides scheduling tools
- Tools: `schedule_task`, `list_tasks`, `pause_task`, `resume_task`, `cancel_task`, `send_message`
- `ncl tasks` provides scheduling commands
- Commands: `list`, `get`, `create`, `update`, `cancel`, `pause`, `resume`, `delete`, `run`, `append-log`
- Tasks stored in SQLite with run history
- Scheduler loop checks for due tasks every minute
- Tasks execute Claude Agent SDK in containerized group context
- Tasks execute in the agent group's system session
### Web Access
- Built-in WebSearch and WebFetch tools
+3 -20
View File
@@ -579,12 +579,7 @@ NanoClaw has a built-in scheduler that runs tasks as full agents in their group'
```
User: @Andy remind me every Monday at 9am to review the weekly metrics
Claude: [calls mcp__nanoclaw__schedule_task]
{
"prompt": "Send a reminder to review weekly metrics. Be encouraging!",
"schedule_type": "cron",
"schedule_value": "0 9 * * 1"
}
Claude: [runs ncl tasks create --prompt "Send a reminder to review weekly metrics. Be encouraging!" --process-after "2024-02-05T09:00:00" --recurrence "0 9 * * 1"]
Claude: Done! I'll remind you every Monday at 9am.
```
@@ -594,12 +589,7 @@ Claude: Done! I'll remind you every Monday at 9am.
```
User: @Andy at 5pm today, send me a summary of today's emails
Claude: [calls mcp__nanoclaw__schedule_task]
{
"prompt": "Search for today's emails, summarize the important ones, and send the summary to the group.",
"schedule_type": "once",
"schedule_value": "2024-01-31T17:00:00Z"
}
Claude: [runs ncl tasks create --prompt "Search for today's emails, summarize the important ones, and send the summary to the group." --process-after "2024-01-31T17:00:00"]
```
### Managing Tasks
@@ -620,18 +610,11 @@ From main channel:
### NanoClaw MCP (built-in)
The `nanoclaw` MCP server is created dynamically per agent call with the current group's context.
The `nanoclaw` MCP server is created dynamically per agent call with the current group's context. Scheduled task management lives in `ncl tasks`, not MCP.
**Available Tools:**
| Tool | Purpose |
|------|---------|
| `schedule_task` | Schedule a recurring or one-time task |
| `list_tasks` | Show tasks (group's tasks, or all if main) |
| `get_task` | Get task details and run history |
| `update_task` | Modify task prompt or schedule |
| `pause_task` | Pause a task |
| `resume_task` | Resume a paused task |
| `cancel_task` | Delete a task |
| `send_message` | Send a message to the group via its channel |
---
+8 -42
View File
@@ -634,52 +634,18 @@ destination is of type `agent`. `resolveRouting` maps it to a `messages_out` row
`channel_type: 'agent'` and `platform_id` set to the target agent group id; the host
validates the send and routes it into the target session's `inbound.db`.
#### schedule_task
#### ncl tasks
Schedule a one-shot or recurring task.
Schedule, inspect, and modify one-shot or recurring tasks.
```typescript
{
name: 'schedule_task',
params: {
prompt: string, // task prompt
processAfter: string, // ISO timestamp for first run
recurrence?: string, // cron expression (optional)
script?: string, // pre-agent script (optional)
}
}
```bash
ncl tasks create --prompt "..." --process-after "2026-01-15T09:00:00" --recurrence "0 9 * * *"
ncl tasks list
ncl tasks update <series_id> --prompt "..."
ncl tasks cancel <series_id>
```
Implementation: the container can't write host-owned `inbound.db`, so this writes a `messages_out` row with `kind: 'system'` and `action: 'schedule_task'` (`container/agent-runner/src/mcp-tools/scheduling.ts`). During delivery the host's action handler (`src/modules/scheduling/actions.ts``insertTask()` in `src/modules/scheduling/db.ts`) inserts the `kind: 'task'` row into `inbound.db` with `process_after` and optionally `recurrence`. The host sweep picks it up when due.
#### list_tasks
List active scheduled/recurring tasks.
```typescript
{
name: 'list_tasks',
params: {}
}
```
Implementation: a read, not a write — the container may read the read-only `inbound.db` mount directly. Returns one row per series (the live pending/paused occurrence): `SELECT series_id AS id, ... FROM messages_in WHERE kind = 'task' AND status IN ('pending','paused') GROUP BY series_id`. See `container/agent-runner/src/mcp-tools/scheduling.ts`.
#### cancel_task / pause_task / resume_task / update_task
Modify a scheduled task.
```typescript
{
name: 'cancel_task',
params: { taskId: string }
}
// pause_task: set status = 'paused' (new status value for recurring tasks)
// resume_task: set status = 'pending'
// update_task: merge { prompt?, recurrence?, processAfter?, script? } into the live row
```
Implementation: all four are sent as system actions (`messages_out`, `kind: 'system'`, `action: 'cancel_task' | 'pause_task' | 'resume_task' | 'update_task'`) — the container never writes `inbound.db`. The host's handlers in `src/modules/scheduling/actions.ts` apply the change against `inbound.db` via `src/modules/scheduling/db.ts`: cancel/pause/resume flip status on the live row(s); update_task reads current content, merges supplied fields, and writes back. All four match by `(id = ? OR series_id = ?) AND kind='task' AND status IN ('pending','paused')`, so they reach the live next occurrence of a recurring task even when the agent passes the original (now-completed) id.
Implementation: the host writes `messages_in` task rows into the agent group's system session (`thread_id = system:tasks`). The host sweep wakes that system-session container when a task is due. The task agent chooses its destination at fire time by emitting `<message to="name">...</message>` or using `send_message`.
#### create_agent
+1 -1
View File
@@ -32,7 +32,7 @@ flowchart TB
direction TB
PollLoop["Poll Loop<br/>(container/agent-runner)"]
Provider["Agent providers<br/>(claude — the only one registered in trunk;<br/>opencode ships via the /add-opencode skill)"]
MCP["MCP Tools<br/>send_message, send_file, edit_message,<br/>add_reaction, send_card, ask_user_question,<br/>schedule_task, create_agent,<br/>install_packages, add_mcp_server"]
MCP["MCP Tools<br/>send_message, send_file, edit_message,<br/>add_reaction, send_card, ask_user_question,<br/>create_agent, install_packages, add_mcp_server<br/>CLI: ncl tasks"]
Skills["Container Skills<br/>(container/skills/)"]
InDB[("inbound.db<br/>host writes<br/>even seq<br/>messages_in<br/>delivered<br/>destinations<br/>session_routing")]
OutDB[("outbound.db<br/>container writes<br/>odd seq<br/>messages_out<br/>processing_ack<br/>session_state<br/>container_state<br/>heartbeat file")]
+4 -7
View File
@@ -904,13 +904,10 @@ MCP tools write to the container's own `outbound.db`. Anything that needs a chan
(There is no `send_to_agent` tool — agent-to-agent is `send_message` to an `agent` destination.)
**Scheduling** (all emit `kind: 'system'` actions except the read-only `list_tasks`):
| Tool | What it does |
|------|-------------|
| `schedule_task` | `action: 'schedule_task'`; host inserts the `kind: 'task'` `messages_in` row with `process_after` + optional `recurrence` |
| `list_tasks` | Read `inbound.db` (read-only) — one row per series: `kind = 'task' AND status IN ('pending','paused') GROUP BY series_id` |
| `pause_task` / `resume_task` / `cancel_task` / `update_task` | matching `action`; host updates the live `messages_in` row(s) |
**Scheduling**: scheduled-task management is not an MCP surface — it lives on
`ncl tasks` (create/list/get/update/cancel/pause/resume/run/append-log). Due
task rows live in the agent group's system session and are woken by the host
sweep.
**Central-DB / self-modification** (`kind: 'system'` actions; host authorizes, often via admin approval):
+52
View File
@@ -0,0 +1,52 @@
# `ncl tasks` migration
## Detect
If an agent mentions `schedule_task`, `list_tasks`, `update_task`, `cancel_task`, `pause_task`, or `resume_task`, it is using the old scheduling MCP surface.
A subtler symptom of a stale container image: the agent reports a task as scheduled, but `ncl tasks list` shows nothing and the host log has `Unknown system action` — the old image's `schedule_task` call is acknowledged in-container and then dropped by the new host. The fix below (rebuild + restart) resolves it.
## Why
Scheduling moved to `ncl tasks`. New tasks are stored in a per-agent-group system session and run there, so a scheduled task does not wake an existing chat session. When it fires, the agent must choose the delivery destination explicitly.
## Fix
Rebuild and restart agent containers so they load the updated MCP tool list and instructions:
```bash
./container/build.sh
launchctl kickstart -k gui/$(id -u)/com.nanoclaw
```
On Linux, restart with `systemctl --user restart nanoclaw`.
Use:
```bash
ncl tasks list
ncl tasks create --group <agent_group_id> --prompt "..." --process-after "2026-01-15T09:00:00" --recurrence "0 9 * * *"
ncl tasks update --id <series_id> --prompt "..."
ncl tasks cancel --id <series_id>
```
## Verify
Run `ncl tasks list`. New task rows should show a system `session_id`, not the chat session that requested the task.
## Legacy tasks (scheduled before this update)
Tasks created through the old MCP tools live in the **chat session** that created them, not in a per-series system session. They are unaffected by this update: they keep firing and delivering exactly as before. Two things to know:
- An agent's own `ncl tasks list` (group scope) shows only its group's task rows; from the **host**, unscoped `ncl tasks list` enumerates everything, and `--session <id>` narrows to one session — that is how you find and manage legacy rows (`ncl tasks cancel --session <chat_session_id> --all` to clear a chat session's tasks).
- The `messages_in` status enum now includes `cancelled` (cancel marks the row and clears its recurrence rather than deleting it). Custom code that exhaustively switches on task status needs the new arm.
## Rollback
Order matters:
1. Remove tasks created through `ncl tasks` (`ncl tasks list` / `delete`) — they live in per-series system sessions the old code doesn't know about.
2. **Wait one sweep (≤60s)** so the host closes the now-empty task sessions.
3. Then revert the update and rebuild the container image.
Reverting before the task sessions are collected leaves system sessions behind that the old `findSessionByAgentGroup` (which has no system-session exclusion) can resolve as the group's session — mis-routing agent-to-agent messages into a dead task thread.
+26 -1
View File
@@ -15,7 +15,7 @@ vi.mock('./log.js', () => ({
}));
import { composeGroupClaudeMd } from './claude-md-compose.js';
import { ensureContainerConfig } from './db/container-configs.js';
import { ensureContainerConfig, updateContainerConfigScalars } from './db/container-configs.js';
import { closeDb, createAgentGroup, initTestDb, runMigrations } from './db/index.js';
import { PERSONA_PREPEND_FILE } from './group-persona.js';
import type { AgentGroup } from './types.js';
@@ -91,3 +91,28 @@ describe('composeGroupClaudeMd persona prepend', () => {
expect(fs.existsSync(path.join(GROUPS_DIR, ag.folder, '.claude-fragments', 'persona.md'))).toBe(false);
});
});
describe('composeGroupClaudeMd scheduling instructions (ncl tasks reach-in)', () => {
// Red-on-delete guard for the `scheduling`/`cli` exclusion at the
// module-fragment loop: the agent is taught `ncl tasks` iff it has ncl.
it('imports module-scheduling.md at the default cli_scope', () => {
const ag = group('ag-sched', 'sched-group');
seed(ag);
composeGroupClaudeMd(ag);
expect(importsOf(ag.folder)).toContain('@./.claude-fragments/module-scheduling.md');
});
it('excludes module-scheduling.md (and module-cli.md) when cli_scope is disabled', () => {
const ag = group('ag-sched-off', 'sched-group-off');
seed(ag);
updateContainerConfigScalars(ag.id, { cli_scope: 'disabled' });
composeGroupClaudeMd(ag);
const imports = importsOf(ag.folder);
expect(imports).not.toContain('@./.claude-fragments/module-scheduling.md');
expect(imports).not.toContain('@./.claude-fragments/module-cli.md');
});
});
+6 -4
View File
@@ -80,10 +80,12 @@ export function composeGroupClaudeMd(group: AgentGroup): void {
}
}
// Built-in module fragments — every MCP tool source file that ships a
// Built-in module fragments — every MCP/CLI module that ships a
// sibling `<name>.instructions.md`. These describe how the agent should
// use that module's MCP tools (schedule_task, install_packages, etc.).
// Skip cli.instructions.md when cli_scope is disabled.
// use that module's tools (`ncl tasks`, install_packages, etc.).
// Skip ncl-dependent instructions when cli_scope is disabled. `scheduling`
// teaches `ncl tasks`, so it is just as dead as `cli` itself when the agent
// has no ncl — dispatch rejects every cli_request and ncl is excluded.
const cliDisabled = configRow?.cli_scope === 'disabled';
const mcpToolsHostDir = path.join(process.cwd(), MCP_TOOLS_HOST_SUBPATH);
if (fs.existsSync(mcpToolsHostDir)) {
@@ -91,7 +93,7 @@ export function composeGroupClaudeMd(group: AgentGroup): void {
const match = entry.match(/^(.+)\.instructions\.md$/);
if (!match) continue;
const moduleName = match[1];
if (moduleName === 'cli' && cliDisabled) continue;
if ((moduleName === 'cli' || moduleName === 'scheduling') && cliDisabled) continue;
desired.set(`module-${moduleName}.md`, {
type: 'symlink',
content: `${SHARED_MCP_TOOLS_CONTAINER_BASE}/${entry}`,
+23
View File
@@ -115,6 +115,15 @@ register({
handler: async (args) => ({ echo: args }),
});
register({
name: 'tasks-list',
description: 'test command (tasks resource)',
resource: 'tasks',
access: 'open',
parseArgs: (raw) => raw,
handler: async (args) => ({ echo: args }),
});
register({
name: 'wirings-list',
description: 'test command (wirings resource — not allowed)',
@@ -218,6 +227,7 @@ beforeEach(() => {
sessions: 'agent_group_id',
destinations: 'agent_group_id',
members: 'agent_group_id',
tasks: 'agent_group_id',
};
mockGetResource.mockImplementation((plural: string) =>
scopeFields[plural] ? { scopeField: scopeFields[plural] } : undefined,
@@ -364,6 +374,19 @@ describe('CLI scope enforcement', () => {
}
});
it('group: allows tasks, auto-fills --group', async () => {
mockGetContainerConfig.mockReturnValue({ cli_scope: 'group' });
const resp = await dispatch({ id: '1', command: 'tasks-list', args: {} }, agentCtx());
expect(resp.ok).toBe(true);
if (resp.ok) {
const data = resp.data as { echo: Record<string, unknown> };
expect(data.echo.group).toBe('g1');
expect(data.echo.id).toBeUndefined();
}
});
it('group: blocks non-whitelisted resources (wirings)', async () => {
mockGetContainerConfig.mockReturnValue({ cli_scope: 'group' });
+84
View File
@@ -0,0 +1,84 @@
/**
* Compact aligned-table renderer for `ncl tasks list` (human mode only).
*
* A task series is a recurring job; each fire is a run. This surfaces the run history
* the raw row hides run count, last/next fire, schedule as an aligned table.
* Driven by the enriched rows from listTasks (tasks.ts); the --json path is
* untouched. `now` is injectable so the relative times are testable.
*/
interface TaskListRow {
series_id: string;
schedule?: string | null;
runs?: number;
failed_runs?: number;
last_run?: string | null;
next_run?: string | null;
status?: string;
log?: string | null;
created_at?: string | null;
prompt?: string | null;
}
const COLS = ['SERIES', 'SCHEDULE', 'RUNS', 'FAILED', 'LAST RUN', 'NEXT RUN', 'STATUS', 'AGE', 'PROMPT'] as const;
function parseMs(iso: string): number {
return Date.parse(/[Z+]|[+-]\d\d:\d\d$/.test(iso) ? iso : iso + 'Z');
}
/** "1m ago" / "in 30s" — coarse, human relative time. */
function duration(ms: number): string {
const s = Math.abs(ms) / 1000;
if (s < 60) return `${Math.round(s)}s`;
if (s < 3600) return `${Math.round(s / 60)}m`;
if (s < 86400) return `${Math.round(s / 3600)}h`;
return `${Math.round(s / 86400)}d`;
}
function lastRun(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
if (Number.isNaN(t)) return iso;
return `${duration(now - t)} ago`;
}
function nextRun(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
if (Number.isNaN(t)) return iso;
return t <= now ? 'due' : `in ${duration(t - now)}`;
}
/** AGE — how long since the series was created. */
function age(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
return Number.isNaN(t) ? '-' : duration(now - t);
}
function clip(s: string | null | undefined, n: number): string {
const v = (s ?? '').replace(/\s+/g, ' ').trim();
return v.length > n ? v.slice(0, n - 1) + '…' : v;
}
export function formatTasksTable(rows: TaskListRow[], now: number = Date.now()): string {
if (!rows.length) return 'No tasks.';
const body = rows.map((r) => [
r.series_id, // full id, copy-pasteable into `ncl tasks get --id <…>`
r.schedule || 'once',
String(r.runs ?? 0),
String(r.failed_runs ?? 0),
lastRun(r.last_run, now),
nextRun(r.next_run, now),
r.status ?? '-',
age(r.created_at, now),
clip(r.prompt, 40),
]);
const widths = COLS.map((c, i) => Math.max(c.length, ...body.map((row) => row[i].length)));
const line = (cells: string[]) =>
cells
.map((c, i) => c.padEnd(widths[i]))
.join(' ')
.trimEnd();
return [line([...COLS]), ...body.map(line)].join('\n');
}
+1 -1
View File
@@ -15,7 +15,7 @@ import type { CallerContext } from './frame.js';
* consumed by both dispatch enforcement and `ncl help` filtering, so the
* agent is never shown a resource the gate would reject (or vice versa).
*/
export const GROUP_SCOPE_RESOURCES = new Set(['groups', 'sessions', 'destinations', 'members']);
export const GROUP_SCOPE_RESOURCES = new Set(['groups', 'sessions', 'destinations', 'members', 'tasks']);
export type Access = 'open' | 'approval' | 'hidden';
+30 -1
View File
@@ -58,10 +58,39 @@ registerResource({
type: 'string',
description: "The target's ID — messaging_groups.id for channels, agent_groups.id for agents.",
},
{ name: 'channel_type', type: 'string', description: 'Resolved channel type for channel destinations.' },
{ name: 'display_name', type: 'string', description: 'Resolved chat title or agent name.' },
{ name: 'created_at', type: 'string', description: 'Auto-set.' },
],
operations: { list: 'open' },
operations: {},
customOperations: {
list: {
access: 'open',
description: 'List destinations with resolved channel/title labels.',
handler: async (args) => {
const agentGroupId = (args.agent_group_id as string | undefined) ?? (args.id as string | undefined);
const params: unknown[] = [];
const where = agentGroupId ? 'WHERE ad.agent_group_id = ?' : '';
if (agentGroupId) params.push(agentGroupId);
return getDb()
.prepare(
`SELECT
ad.agent_group_id,
ad.local_name,
ad.target_type,
ad.target_id,
CASE WHEN ad.target_type = 'channel' THEN mg.channel_type ELSE NULL END AS channel_type,
CASE WHEN ad.target_type = 'channel' THEN mg.name ELSE ag.name END AS display_name,
ad.created_at
FROM agent_destinations ad
LEFT JOIN messaging_groups mg ON ad.target_type = 'channel' AND ad.target_id = mg.id
LEFT JOIN agent_groups ag ON ad.target_type = 'agent' AND ad.target_id = ag.id
${where}
ORDER BY ad.agent_group_id, ad.local_name`,
)
.all(...params);
},
},
add: {
access: 'approval',
description: 'Add a destination for an agent. Use --agent-group-id, --local-name, --target-type, --target-id.',
+1
View File
@@ -14,3 +14,4 @@ import './user-dms.js';
import './dropped-messages.js';
import './approvals.js';
import './sessions.js';
import './tasks.js';
+575
View File
@@ -0,0 +1,575 @@
import Database from 'better-sqlite3';
import fs from 'fs';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('../../config.js', async () => {
const actual = await vi.importActual('../../config.js');
return {
...actual,
DATA_DIR: '/tmp/nanoclaw-test-cli-tasks',
GROUPS_DIR: '/tmp/nanoclaw-test-cli-tasks/groups',
TIMEZONE: 'UTC',
};
});
vi.mock('../../container-runner.js', () => ({
wakeContainer: vi.fn().mockResolvedValue(undefined),
isContainerRunning: vi.fn().mockReturnValue(false),
getActiveContainerCount: vi.fn().mockReturnValue(0),
killContainer: vi.fn(),
}));
const TEST_DIR = '/tmp/nanoclaw-test-cli-tasks';
import { initTestDb, closeDb, runMigrations, createAgentGroup } from '../../db/index.js';
import { createSession, findSessionByAgentGroup, getSessionsByAgentGroup, taskThreadId } from '../../db/sessions.js';
import { countDueMessages } from '../../db/session-db.js';
import { inboundDbPath, initSessionFolder } from '../../session-manager.js';
import { dispatch } from '../dispatch.js';
import { formatTasksTable } from '../format-tasks.js';
import type { CallerContext } from '../frame.js';
import './tasks.js';
import '../commands/index.js'; // registers tasks-help for the help-topic test
function now(): string {
return new Date().toISOString();
}
function createGroup(id: string): void {
createAgentGroup({ id, name: id, folder: id, agent_provider: null, created_at: now() });
}
function createChatSession(group: string, id: string): void {
createSession({
id,
agent_group_id: group,
messaging_group_id: null,
thread_id: null,
agent_provider: null,
status: 'active',
container_status: 'stopped',
last_active: null,
created_at: now(),
});
initSessionFolder(group, id);
}
function agentCtx(group = 'ag-1', session = 'chat-1'): CallerContext {
return { caller: 'agent', agentGroupId: group, sessionId: session, messagingGroupId: 'mg-1' };
}
describe('tasks CLI resource', () => {
beforeEach(() => {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
fs.mkdirSync(TEST_DIR, { recursive: true });
const db = initTestDb();
runMigrations(db);
createGroup('ag-1');
createGroup('ag-2');
createChatSession('ag-1', 'chat-1');
createChatSession('ag-2', 'chat-2');
});
afterEach(() => {
closeDb();
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
});
it('create writes the task into the group system session, not the caller chat session', async () => {
const resp = await dispatch(
{
id: 'req-1',
command: 'tasks-create',
args: { prompt: 'send a briefing', process_after: '2026-01-15T09:00:00Z' },
},
agentCtx(),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
const created = resp.data as { series_id: string; session_id: string };
expect(created.session_id).not.toBe('chat-1');
// The task lands in its own isolated per-series session, not the chat session.
const sessions = getSessionsByAgentGroup('ag-1');
const taskSession = sessions.find((s) => s.id === created.session_id);
expect(taskSession?.thread_id).toBe(taskThreadId(created.series_id));
const chatDb = new Database(inboundDbPath('ag-1', 'chat-1'), { readonly: true });
expect(chatDb.prepare("SELECT COUNT(*) AS count FROM messages_in WHERE kind = 'task'").get()).toEqual({
count: 0,
});
chatDb.close();
const systemDb = new Database(inboundDbPath('ag-1', created.session_id), { readonly: true });
const row = systemDb.prepare("SELECT content FROM messages_in WHERE kind = 'task'").get() as { content: string };
const content = JSON.parse(row.content);
expect(content).toMatchObject({ originSessionId: 'chat-1' });
expect(content.prompt).toContain('send a briefing');
expect(content.prompt).toContain(`tasks/${created.series_id}.md`); // log-path hint injected
systemDb.close();
});
it('tasks-list attaches a server-rendered human table (so the container agent gets it too)', async () => {
await dispatch(
{
id: 'c',
command: 'tasks-create',
args: { prompt: 'x', name: 'briefing', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx(),
);
const resp = await dispatch({ id: 'l', command: 'tasks-list', args: {} }, agentCtx());
expect(resp.ok).toBe(true);
if (!resp.ok) return;
// Red-on-delete guard for the dispatch wiring: the host renders format-tasks
// once and ships it as `human`, so the Bun container prints the aligned
// table instead of a raw column dump (it cannot import the host formatter).
expect(resp.human).toBeDefined();
expect(resp.human).toMatch(/SERIES\s+SCHEDULE\s+RUNS\s+FAILED\s+LAST RUN\s+NEXT RUN/);
expect(resp.human).toContain('briefing-');
});
it('recurrence more frequent than 4x/day is refused with the quota warning', async () => {
const resp = await dispatch(
{ id: 'c', command: 'tasks-create', args: { prompt: 'x', name: 'spam', recurrence: '*/2 * * * *' } },
agentCtx(),
);
expect(resp.ok).toBe(false);
if (!resp.ok) {
expect(resp.error.message).toContain('this task has not been scheduled');
expect(resp.error.message).toContain('ncl tasks create --help');
expect(resp.error.message).toContain('--dangerously-override-recurrence-limit');
}
});
it('exactly 4 fires/day passes; the override flag bypasses the limit', async () => {
const four = await dispatch(
{ id: 'c4', command: 'tasks-create', args: { prompt: 'x', name: 'four', recurrence: '0 0,6,12,18 * * *' } },
agentCtx(),
);
expect(four.ok).toBe(true);
const overridden = await dispatch(
{
id: 'co',
command: 'tasks-create',
args: { prompt: 'x', name: 'fast', recurrence: '*/30 * * * *', dangerously_override_recurrence_limit: true },
},
agentCtx(),
);
expect(overridden.ok).toBe(true);
});
it('a --script gate exempts frequent recurrence — the sanctioned monitor pattern', async () => {
const scripted = await dispatch(
{
id: 'cs',
command: 'tasks-create',
args: {
prompt: 'triage queue',
name: 'watch',
recurrence: '*/10 * * * *',
script: 'echo {"wakeAgent": false}',
},
},
agentCtx(),
);
expect(scripted.ok).toBe(true);
// update --recurrence on a task that already has a script: also exempt.
if (!scripted.ok) return;
const seriesId = (scripted.data as { series_id: string }).series_id;
const upd = await dispatch(
{ id: 'us', command: 'tasks-update', args: { id: seriesId, recurrence: '*/5 * * * *' } },
agentCtx(),
);
expect(upd.ok).toBe(true);
// …but clearing the script in the same update re-arms the guard.
const cleared = await dispatch(
{ id: 'uc', command: 'tasks-update', args: { id: seriesId, recurrence: '*/5 * * * *', script: 'none' } },
agentCtx(),
);
expect(cleared.ok).toBe(false);
});
it('the limit also guards update --recurrence (no create-slow-then-update bypass)', async () => {
const created = await dispatch(
{ id: 'c', command: 'tasks-create', args: { prompt: 'x', name: 'sneak', recurrence: '0 9 * * *' } },
agentCtx(),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const seriesId = (created.data as { series_id: string }).series_id;
const upd = await dispatch(
{ id: 'u', command: 'tasks-update', args: { id: seriesId, recurrence: '* * * * *' } },
agentCtx(),
);
expect(upd.ok).toBe(false);
if (!upd.ok) expect(upd.error.message).toContain('this task has not been scheduled');
});
it('tasks create --help carries the script contract and the frequency-limit caveat', async () => {
// --help and `tasks help create` render the same deep verb help.
const resp = await dispatch({ id: 'h', command: 'tasks-create', args: { help: true } }, agentCtx());
expect(resp.ok).toBe(true);
if (resp.ok) {
const text = resp.data as string;
expect(text).toContain('wakeAgent');
expect(text).toContain('Frequency limit');
}
});
it('agent-shared lookup skips the task system session', async () => {
await dispatch(
{
id: 'req-1',
command: 'tasks-create',
args: { prompt: 'send a briefing', process_after: '2026-01-15T09:00:00Z' },
},
agentCtx(),
);
expect(findSessionByAgentGroup('ag-1')?.id).toBe('chat-1');
});
it('group-scoped agents cannot list tasks from another group session', async () => {
const resp = await dispatch(
{ id: 'req-1', command: 'tasks-list', args: { session: 'chat-2' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) {
expect(resp.error.code).toBe('handler-error');
expect(resp.error.message).toContain('session not found');
}
});
it('--name yields a short, readable, fs/thread-safe id', async () => {
const r = await dispatch(
{
id: 'rn',
command: 'tasks-create',
args: { prompt: 'x', name: 'Morning Joke!!', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const id = (r.data as { series_id: string }).series_id;
expect(id).toMatch(/^morning-joke-[0-9a-f]{4}$/);
expect(id).toMatch(/^[a-z0-9-]+$/); // safe as thread suffix / filename / --id
});
it('no name yields a t-<hex> id', async () => {
const r = await dispatch(
{ id: 'rnn', command: 'tasks-create', args: { prompt: 'x', process_after: '2999-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
expect((r.data as { series_id: string }).series_id).toMatch(/^t-[0-9a-f]{6}$/);
});
it('recurring create derives the first run from the cron grid when --process-after is omitted', async () => {
const r = await dispatch(
{ id: 'rec', command: 'tasks-create', args: { prompt: 'x', name: 'nightly', recurrence: '0 9 * * 1-5' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const task = r.data as { process_after: string; recurrence: string };
expect(task.recurrence).toBe('0 9 * * 1-5');
// First fire snapped onto the cron grid (TIMEZONE=UTC in this suite).
const firstRun = new Date(task.process_after);
expect(Number.isNaN(firstRun.getTime())).toBe(false);
expect(firstRun.getUTCHours()).toBe(9);
expect(firstRun.getTime()).toBeGreaterThan(Date.now());
});
it('one-shot create still requires --process-after (nothing to derive it from)', async () => {
const r = await dispatch(
{ id: 'os', command: 'tasks-create', args: { prompt: 'x', name: 'once' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(false);
if (!r.ok) expect(r.error.message).toContain('--process-after is required');
});
it('run queues an extra immediate occurrence without consuming the scheduled one', async () => {
const created = await dispatch(
{
id: 'c',
command: 'tasks-create',
args: { prompt: 'x', name: 'pingable', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { series_id, session_id } = created.data as { series_id: string; session_id: string };
const run = await dispatch({ id: 'r', command: 'tasks-run', args: { id: series_id } }, agentCtx('ag-1', 'chat-1'));
expect(run.ok).toBe(true);
if (!run.ok) return;
const fired = run.data as { series_id: string; row_id: string; status: string };
expect(fired.series_id).toBe(series_id);
expect(fired.row_id).not.toBe(series_id);
expect(fired.status).toBe('pending');
const db = new Database(inboundDbPath('ag-1', session_id), { readonly: true });
const pending = db
.prepare(
"SELECT id, recurrence, process_after FROM messages_in WHERE kind = 'task' AND status = 'pending' AND series_id = ?",
)
.all(series_id) as Array<{ id: string; recurrence: string | null; process_after: string }>;
db.close();
// Original scheduled row + the new run-now occurrence both still pending.
expect(pending).toHaveLength(2);
const runRow = pending.find((p) => p.id === fired.row_id);
expect(runRow?.recurrence).toBeNull(); // never re-armed into a phantom series
expect(new Date(runRow!.process_after).getTime()).toBeLessThanOrEqual(Date.now());
});
it('task object exposes origin_session_id and created_at', async () => {
const r = await dispatch(
{
id: 'ro',
command: 'tasks-create',
args: { prompt: 'x', name: 'o', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const d = r.data as { origin_session_id: string | null; created_at: string };
expect(d.origin_session_id).toBe('chat-1'); // the session that created it
expect(d.created_at).toBeTruthy();
});
it('each task gets its own isolated session, and list fans out across them', async () => {
const a = await dispatch(
{ id: 'r-a', command: 'tasks-create', args: { prompt: 'task A', process_after: '2026-01-15T09:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
const b = await dispatch(
{ id: 'r-b', command: 'tasks-create', args: { prompt: 'task B', process_after: '2026-01-15T09:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(a.ok && b.ok).toBe(true);
if (!a.ok || !b.ok) return;
const ta = a.data as { session_id: string; series_id: string };
const tb = b.data as { session_id: string; series_id: string };
// Distinct per-series sessions — not one shared system:tasks session.
expect(ta.session_id).not.toBe(tb.session_id);
// list (no --session) fans out across every task session in the group.
const list = await dispatch({ id: 'r-l', command: 'tasks-list', args: {} }, agentCtx('ag-1', 'chat-1'));
expect(list.ok).toBe(true);
if (!list.ok) return;
const ids = (list.data as Array<{ series_id: string }>).map((t) => t.series_id);
expect(ids).toContain(ta.series_id);
expect(ids).toContain(tb.series_id);
});
it('list enriches each series with run history (CronJob view)', async () => {
const created = await dispatch(
{
id: 'r-agg',
command: 'tasks-create',
args: { prompt: 'brain digest', recurrence: '0 9 * * *', 'process-after': '2026-01-15T09:05:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { session_id, series_id } = created.data as { session_id: string; series_id: string };
// Seed three completed fires for this series into its own session inbound.db.
const db = new Database(inboundDbPath('ag-1', session_id));
const ins = db.prepare(
'INSERT INTO messages_in (id, seq, timestamp, status, tries, kind, content, series_id, process_after) ' +
"VALUES (?, ?, datetime('now'), 'completed', 0, 'task', '{}', ?, ?)",
);
ins.run('run-1', 100, series_id, '2026-01-15T09:02:00Z');
ins.run('run-2', 102, series_id, '2026-01-15T09:03:00Z');
ins.run('run-3', 104, series_id, '2026-01-15T09:04:00Z');
db.close();
const list = await dispatch({ id: 'r-agg-l', command: 'tasks-list', args: {} }, agentCtx('ag-1', 'chat-1'));
expect(list.ok).toBe(true);
if (!list.ok) return;
const row = (list.data as Array<Record<string, unknown>>).find((t) => t.series_id === series_id);
expect(row).toBeDefined();
expect(row?.runs).toBe(3);
expect(row?.last_run).toBe('2026-01-15T09:04:00Z'); // max completed process_after
expect(String(row?.next_run)).toMatch(/^2026-01-15T09:05:00/); // the live pending occurrence
expect(row?.schedule).toBe('0 9 * * *');
expect(row?.log).toBe(`tasks/${series_id}.md`);
});
// The schedule→wake primitive without a container: a task created through the
// real `ncl tasks create` path must land in the agent group's system session
// AND be counted by the same due-message query the host sweep uses to decide a
// wake. Goes red if trigger defaulting, system-session routing, or the due
// predicate ever drift apart.
describe('a due task makes the system session wakeable', () => {
it('countDueMessages sees a past task and ignores a future one', async () => {
const created = await dispatch(
{ id: 'r-due', command: 'tasks-create', args: { prompt: 'run me', 'process-after': '2020-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const systemId = (created.data as { session_id: string }).session_id;
const dueDb = new Database(inboundDbPath('ag-1', systemId), { readonly: true });
expect(countDueMessages(dueDb)).toBe(1); // host sweep would wake this session
dueDb.close();
// A far-future task in the same system session is not yet due.
const future = await dispatch(
{ id: 'r-fut', command: 'tasks-create', args: { prompt: 'later', 'process-after': '2999-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(future.ok).toBe(true);
const stillDb = new Database(inboundDbPath('ag-1', systemId), { readonly: true });
expect(countDueMessages(stillDb)).toBe(1); // still just the one past task
stillDb.close();
});
});
describe('append-log', () => {
const logFile = (folder: string, series: string) => `${TEST_DIR}/groups/${folder}/tasks/${series}.md`;
it('writes a host-timestamped line to the run log and creates the file (explicit --id)', async () => {
const resp = await dispatch(
{ id: 'al-1', command: 'tasks-append-log', args: { id: 'my-task-1', msg: 'did the thing; it worked' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
const content = fs.readFileSync(logFile('ag-1', 'my-task-1'), 'utf8').trim();
expect(content).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z — did the thing; it worked$/);
});
it('derives the series from the caller task session when --id is omitted', async () => {
const created = await dispatch(
{
id: 'al-c',
command: 'tasks-create',
args: { name: 'derive-me', prompt: 'x', 'process-after': '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { series_id, session_id } = created.data as { series_id: string; session_id: string };
// The fire runs INSIDE that task session, so no --id is needed.
const resp = await dispatch(
{ id: 'al-2', command: 'tasks-append-log', args: { msg: 'auto-derived run' } },
agentCtx('ag-1', session_id),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
expect((resp.data as { series: string }).series).toBe(series_id);
expect(fs.readFileSync(logFile('ag-1', series_id), 'utf8')).toContain('auto-derived run');
});
it('requires --msg', async () => {
const resp = await dispatch(
{ id: 'al-3', command: 'tasks-append-log', args: { id: 'my-task-1' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) expect(resp.error.message).toContain('--msg is required');
});
it('errors when there is no --id and the caller is not in a task session', async () => {
// chat-1 is a normal chat session, not system:tasks:* → nothing to derive.
const resp = await dispatch(
{ id: 'al-4', command: 'tasks-append-log', args: { msg: 'orphan' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) expect(resp.error.message).toMatch(/--id is required/);
});
});
});
describe('formatTasksTable', () => {
const now = Date.parse('2026-01-15T09:05:30Z');
const rows = [
{
series_id: 'task-5bbe082a-6298-4699',
schedule: '* * * * *',
runs: 7,
failed_runs: 2,
last_run: '2026-01-15T09:04:30Z',
next_run: '2026-01-15T09:06:00Z',
status: 'pending',
log: 'tasks/task-5bbe082a.md',
created_at: '2026-01-15T08:05:30Z', // 1h before now
prompt: 'You are NanoClaw, wired into the company brain, your job this run is to read it',
},
];
it('renders an aligned table with run history', () => {
const lines = formatTasksTable(rows, now).split('\n');
expect(lines[0]).toMatch(/SERIES\s+SCHEDULE\s+RUNS\s+FAILED\s+LAST RUN\s+NEXT RUN\s+STATUS\s+AGE\s+PROMPT/);
expect(lines[1]).toContain('1h'); // AGE column — created 1h ago
expect(lines[1]).toContain('task-5bbe082a-6298-4699'); // FULL series id — copy-pasteable into `tasks get --id`
expect(lines[1]).toContain('* * * * *');
expect(lines[1]).toContain('1m ago'); // 09:04:30 vs 09:05:30
expect(lines[1]).toContain('in 30s'); // 09:06:00 vs 09:05:30
expect(lines[1]).toContain('…'); // prompt truncated
});
it('handles a never-fired series and an empty list', () => {
expect(formatTasksTable([], now)).toBe('No tasks.');
const oneShot = formatTasksTable(
[
{
series_id: 'task-x',
schedule: 'once',
runs: 0,
last_run: null,
next_run: '2026-01-15T09:00:00Z',
status: 'pending',
},
],
now,
).split('\n')[1];
expect(oneShot).toContain('once');
expect(oneShot).toMatch(/\bdue\b/); // next_run in the past → due
expect(oneShot).toContain('-'); // last_run '-' (never fired)
});
});
describe('deep verb help (ncl tasks help create)', () => {
it('resolves through the dispatcher fallback and renders the full contract + examples', async () => {
// Side-effect import mirrors the CLI server boot: registers <plural>-help.
await import('../commands/index.js');
const resp = await dispatch({ id: 'h1', command: 'tasks-help-create', args: {} }, { caller: 'host' });
expect(resp.ok).toBe(true);
if (resp.ok) {
const text = resp.data as string;
expect(text).toContain('ncl tasks create');
expect(text).toContain('wakeAgent'); // full multi-line script contract present
expect(text).toContain('Examples:'); // examples block rendered
}
});
it('rejects an unknown verb with a pointer back to resource help', async () => {
await import('../commands/index.js');
const resp = await dispatch({ id: 'h2', command: 'tasks-help-frobnicate', args: {} }, { caller: 'host' });
expect(resp.ok).toBe(false);
});
});
+784
View File
@@ -0,0 +1,784 @@
import { randomUUID } from 'crypto';
import fs from 'fs';
import type Database from 'better-sqlite3';
import { CronExpressionParser } from 'cron-parser';
import { GROUPS_DIR, TIMEZONE } from '../../config.js';
import { getAgentGroup } from '../../db/agent-groups.js';
import {
findTaskSessions,
getActiveSessions,
getSession,
isTaskThread,
TASKS_SYSTEM_THREAD_ID,
} from '../../db/sessions.js';
import {
cancelAllTasks,
cancelTask,
deleteTask,
insertTaskRow,
pauseTask,
resumeTask,
updateTask,
type TaskUpdate,
} from '../../modules/scheduling/db.js';
import { inboundDbPath, resolveTaskSession, withInboundDb } from '../../session-manager.js';
import { parseZonedToUtc } from '../../timezone.js';
import { registerResource } from '../crud.js';
import { formatTasksTable } from '../format-tasks.js';
import type { CallerContext } from '../frame.js';
type TaskStatus = 'pending' | 'paused';
interface TaskRow {
row_id: string;
series_id: string | null;
status: string;
process_after: string | null;
recurrence: string | null;
content: string;
timestamp: string;
tries: number;
seq: number;
}
interface ScopedSession {
id: string;
agent_group_id: string;
}
function str(value: unknown): string | undefined {
return typeof value === 'string' && value.length > 0 ? value : undefined;
}
function bool(value: unknown): boolean {
return value === true || value === 'true' || value === '1';
}
/**
* Short, readable, filesystem/thread-safe task id. With a name `<slug>-<4hex>`
* (e.g. "Morning joke" `morning-joke-a25c`); without `t-<6hex>`. Always
* matches /^[a-z0-9-]+$/ so it is safe as a thread suffix (`system:tasks:<id>`),
* a filename (`tasks/<id>.md`), and a copy-pasteable --id.
*/
function makeTaskId(name: unknown): string {
const hex = (n: number): string => randomUUID().replace(/-/g, '').slice(0, n);
const slug =
typeof name === 'string'
? name
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.slice(0, 24)
.replace(/-+$/g, '')
: '';
return slug ? `${slug}-${hex(4)}` : `t-${hex(6)}`;
}
function parseProcessAfter(value: unknown): string {
const raw = str(value);
if (!raw) throw new Error('--process-after is required');
const date = parseZonedToUtc(raw, TIMEZONE);
if (Number.isNaN(date.getTime())) throw new Error(`invalid --process-after: ${raw}`);
return date.toISOString();
}
/**
* First-run timestamp for a new task. When a recurrence is given but no
* --process-after, derive the first fire from the cron grid (in TIMEZONE) so the
* common recurring case is a single flag `--recurrence "0 9 * * 1-5"` with no
* redundant, easily-stale hand-picked instant. --process-after is still required
* for one-shots (no recurrence to derive from) and still wins when supplied.
*/
function firstRunIso(value: unknown, recurrence: string | null): string {
if (str(value) === undefined && recurrence) {
const next = CronExpressionParser.parse(recurrence, { tz: TIMEZONE }).next().toISOString();
if (!next) throw new Error(`--recurrence has no upcoming run: ${recurrence}`);
return next;
}
return parseProcessAfter(value);
}
function normalizeNullableString(value: unknown): string | null | undefined {
if (value === undefined) return undefined;
if (value === null) return null;
if (typeof value !== 'string') return String(value);
const trimmed = value.trim();
if (trimmed === '' || trimmed === 'null' || trimmed === 'none') return null;
return value;
}
function validateRecurrence(value: string | null | undefined): void {
if (!value) return;
try {
CronExpressionParser.parse(value, { tz: TIMEZONE });
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
throw new Error(`invalid --recurrence: ${msg}`, { cause: err });
}
}
/**
* Frequency guard: refuse recurrences more frequent than 4 fires/day unless
* the agent explicitly overrides. Frequent tasks burn the user's quota (or
* get their account banned) the sanctioned pattern is a slower cron plus a
* pre-task gate script that checks an external condition and only wakes the
* agent when something changed (`ncl tasks create --help`). Counted over the
* next 24h from now in the instance timezone, so uneven crons are judged by
* what they would actually do.
*/
const MAX_DAILY_FIRES = 4;
const RECURRENCE_LIMIT_WARNING =
'Warning: this task has not been scheduled. Frequent running tasks consume the ' +
"user's subscription quota or unnecessarily use tokens and can cause the user's " +
'account to be banned. Instead, use a pre-task run script that you write that can ' +
'check some kind of external condition, usually via one or more API calls. The ' +
'script returns a decision programmatically whether the task needs to be run now ' +
'or not. For example, an API call to GitHub to check if there are open PRs, and ' +
'only run when there are new open PRs.\n' +
'Run `ncl tasks create --help` to get full directions on how to write a script and test it.\n\n' +
'Note: if and only if you explicitly need to schedule a task more frequently and ' +
"you've verified with the user that they understand and that this is what they " +
'want and based on your judgment you agree that this is the right thing to do in ' +
'this situation, you can override this with --dangerously-override-recurrence-limit';
function enforceRecurrenceLimit(recurrence: string | null, override: boolean, hasScript: boolean): void {
// A gate script IS the sanctioned mitigation the warning steers toward — a
// script-gated fire that finds nothing never wakes the agent, so scripted
// tasks may run at any cadence without the override.
if (!recurrence || override || hasScript) return;
const horizon = Date.now() + 24 * 60 * 60 * 1000;
const interval = CronExpressionParser.parse(recurrence, { tz: TIMEZONE });
let fires = 0;
while (fires <= MAX_DAILY_FIRES) {
const next = interval.next();
if (next.getTime() > horizon) break;
fires++;
}
if (fires > MAX_DAILY_FIRES) throw new Error(RECURRENCE_LIMIT_WARNING);
}
function statusFilter(args: Record<string, unknown>): TaskStatus | undefined {
const status = str(args.status);
if (!status) return undefined;
if (status !== 'pending' && status !== 'paused') {
throw new Error('--status must be pending or paused');
}
return status;
}
function groupArg(args: Record<string, unknown>, ctx: CallerContext): string | undefined {
if (ctx.caller === 'agent') return ctx.agentGroupId;
return str(args.group) ?? str(args.agent_group_id);
}
function ownSession(sessionId: string, ctx: CallerContext): ScopedSession {
const session = getSession(sessionId);
if (!session) throw new Error(`session not found: ${sessionId}`);
if (ctx.caller === 'agent' && session.agent_group_id !== ctx.agentGroupId) {
throw new Error(`session not found: ${sessionId}`);
}
return { id: session.id, agent_group_id: session.agent_group_id };
}
function selectedSessions(args: Record<string, unknown>, ctx: CallerContext): ScopedSession[] {
const sessionId = str(args.session);
if (sessionId) return [ownSession(sessionId, ctx)];
const group = groupArg(args, ctx);
if (group) {
// One session per live task series — the loops below already fan out across them.
return findTaskSessions(group).map((s) => ({ id: s.id, agent_group_id: s.agent_group_id }));
}
if (ctx.caller === 'agent') return [];
return getActiveSessions().map((s) => ({ id: s.id, agent_group_id: s.agent_group_id }));
}
function withInbound<T>(session: ScopedSession, fn: (db: Database.Database) => T): T | undefined {
if (!fs.existsSync(inboundDbPath(session.agent_group_id, session.id))) return undefined;
return withInboundDb(session.agent_group_id, session.id, fn);
}
function parseContent(raw: string): { prompt: string; script: string | null; originSessionId: string | null } {
try {
const parsed = JSON.parse(raw) as Record<string, unknown>;
return {
prompt: typeof parsed.prompt === 'string' ? parsed.prompt : '',
script: typeof parsed.script === 'string' ? parsed.script : null,
originSessionId: typeof parsed.originSessionId === 'string' ? parsed.originSessionId : null,
};
} catch {
// LEGACY-COMPAT(v1-tasks): plain-string content from rows that predate the
// JSON envelope. Removable once no pre-v2 session DBs remain in the wild.
return { prompt: raw, script: null, originSessionId: null };
}
}
function toOutput(session: ScopedSession, row: TaskRow) {
const content = parseContent(row.content);
return {
agent_group_id: session.agent_group_id,
session_id: session.id,
series_id: row.series_id ?? row.row_id,
row_id: row.row_id,
status: row.status,
process_after: row.process_after,
recurrence: row.recurrence,
prompt: content.prompt.length > 120 ? content.prompt.slice(0, 117) + '...' : content.prompt,
has_script: content.script ? 1 : 0,
origin_session_id: content.originSessionId, // which session created the task (null for CLI-created)
created_at: row.timestamp,
tries: row.tries,
};
}
function selectLiveTasks(db: Database.Database, status?: TaskStatus): TaskRow[] {
const statusSql = status ? 'status = ?' : "status IN ('pending', 'paused')";
return db
.prepare(
`SELECT id AS row_id, series_id, status, process_after, recurrence, content, timestamp, tries, MAX(seq) AS seq
FROM messages_in
WHERE kind = 'task'
AND ${statusSql}
GROUP BY series_id
ORDER BY datetime(process_after) ASC, seq ASC`,
)
.all(...(status ? [status] : [])) as TaskRow[];
}
function selectTask(db: Database.Database, id: string): TaskRow | undefined {
return db
.prepare(
`SELECT id AS row_id, series_id, status, process_after, recurrence, content, timestamp, tries, seq
FROM messages_in
WHERE kind = 'task'
AND (id = ? OR series_id = ?)
ORDER BY CASE WHEN status IN ('pending', 'paused') THEN 0 ELSE 1 END, seq DESC
LIMIT 1`,
)
.get(id, id) as TaskRow | undefined;
}
function taskId(args: Record<string, unknown>): string {
const id = str(args.id);
if (!id) throw new Error('task series id is required');
return id;
}
function createTask(args: Record<string, unknown>, ctx: CallerContext) {
const group = groupArg(args, ctx);
if (!group) throw new Error('--group is required');
const prompt = str(args.prompt);
if (!prompt) throw new Error('--prompt is required');
const recurrence = normalizeNullableString(args.recurrence) ?? null;
validateRecurrence(recurrence);
const script = normalizeNullableString(args.script) ?? null;
enforceRecurrenceLimit(recurrence, bool(args.dangerously_override_recurrence_limit), script != null);
const processAfter = firstRunIso(args.process_after, recurrence);
const id = makeTaskId(args.name);
const originSessionId = ctx.caller === 'agent' ? ctx.sessionId : null;
// Each series runs in its own isolated session; point the fire at its own log.
const { session } = resolveTaskSession(group, id);
const promptWithLog =
`${prompt}\n\n` +
`[A task serves the user two separate ways — do whichever the task above asks for, and ALWAYS the run log:\n` +
`• MESSAGE (only if asked): if the task says to report/notify the user, send your result with an EXPLICIT destination — <message to="name">…</message> or send_message({ to: "name", … }). This run has no chat attached: an unaddressed reply is DISCARDED, so the explicit send is the ONLY thing the user receives.\n` +
`• RUN LOG (ALWAYS — even if you sent no message and did nothing else this run): after any sends, end the run with:\n` +
` ncl tasks append-log --msg "<what you did, and why it mattered>"\n` +
` Write it like a work-log entry a human keeps — concrete: what you did and WHY (a no-op run still gets a line saying why nothing was needed). If you wrote or modified files this run, name them in --msg. Not a greeting, not a copy of the message you sent. The host stamps the UTC time (do NOT add one), do NOT edit tasks/${id}.md by hand, and this NEVER goes to the user.\n` +
`Need context from past runs? Read tasks/${id}.md first.]`;
const created = withInbound(session, (db) => {
insertTaskRow(db, {
id,
seriesId: id,
processAfter,
recurrence,
content: JSON.stringify({ prompt: promptWithLog, script, originSessionId }),
});
return selectTask(db, id);
});
if (!created) throw new Error('task system session inbound.db not found');
return toOutput(session, created);
}
/**
* Append one host-timestamped line to a task's run log
* (`<GROUPS_DIR>/<folder>/tasks/<series>.md`). This is NOT a delivery it writes
* nothing to messages_out; it just records what happened so the agent (and human)
* can see when and why each fire ran. Inside a task fire the series is derived from
* the caller's own task session, so the agent supplies only --msg.
*/
function appendTaskLog(
args: Record<string, unknown>,
ctx: CallerContext,
): { series: string; timestamp: string; path: string; ok: true } {
const msg = str(args.msg);
if (!msg) throw new Error('--msg is required');
let series = str(args.id);
let group = groupArg(args, ctx);
if (!series && ctx.caller === 'agent' && ctx.sessionId) {
const sess = getSession(ctx.sessionId);
if (sess && sess.thread_id && isTaskThread(sess.thread_id)) {
series = sess.thread_id.slice(`${TASKS_SYSTEM_THREAD_ID}:`.length);
group ??= sess.agent_group_id;
}
}
if (!series) throw new Error('--id is required (no task session to derive it from)');
// Charset guard is the security boundary here: blocks path traversal and keeps
// the id safe as a filename / thread suffix. Group scope is already enforced by
// groupArg (a cli_scope=group caller can only ever resolve its own folder), so a
// foreign id at worst writes a stray log under the caller's OWN folder — no leak.
if (!/^[a-z0-9-]+$/.test(series)) throw new Error(`invalid task id: ${series}`);
if (!group) throw new Error('could not resolve the agent group');
const ag = getAgentGroup(group);
if (!ag) throw new Error(`agent group not found: ${group}`);
const timestamp = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
const dir = `${GROUPS_DIR}/${ag.folder}/tasks`;
const file = `${dir}/${series}.md`;
fs.mkdirSync(dir, { recursive: true });
fs.appendFileSync(file, `${timestamp}${msg}\n`);
return { series, timestamp, path: file, ok: true };
}
/**
* Run history for one task series, aggregated over its occurrence rows: number
* of successful fires, the last fire time, and failed fires (a row reaches
* `failed` after MAX_TRIES on a stuck claim). Cancelled occurrences are
* `cancelled`, not `completed`, so they never inflate the run count.
*/
function seriesStats(
db: Database.Database,
seriesKey: string,
): { runs: number; last_run: string | null; failed_runs: number } {
return db
.prepare(
`SELECT
COUNT(*) FILTER (WHERE status = 'completed') AS runs,
MAX(process_after) FILTER (WHERE status = 'completed') AS last_run,
COUNT(*) FILTER (WHERE status = 'failed') AS failed_runs
FROM messages_in
WHERE kind = 'task' AND (id = ? OR series_id = ?)`,
)
.get(seriesKey, seriesKey) as { runs: number; last_run: string | null; failed_runs: number };
}
/** Last ~10 lines of a series' run log (`tasks/<series>.md`), newest last. */
function tailRunLog(agentGroupId: string, seriesKey: string, lines = 10): string[] {
const ag = getAgentGroup(agentGroupId);
if (!ag) return [];
const file = `${GROUPS_DIR}/${ag.folder}/tasks/${seriesKey}.md`;
if (!fs.existsSync(file)) return [];
return fs.readFileSync(file, 'utf8').trimEnd().split('\n').filter(Boolean).slice(-lines);
}
/**
* A task series is CronJob-like: the live (pending/paused) row is the next run,
* and the `completed` rows are its run history. Enrich each listed series with
* that history run count, failures, last fire, next fire, schedule, and a
* pointer to the agent's own run log so `tasks list` reads as a compact
* run-history table.
*/
function enrichListRow(db: Database.Database, base: ReturnType<typeof toOutput>) {
const seriesKey = base.series_id;
const stats = seriesStats(db, seriesKey);
return {
...base,
schedule: base.recurrence ?? 'once',
runs: stats.runs,
failed_runs: stats.failed_runs,
last_run: stats.last_run,
next_run: base.process_after,
log: `tasks/${seriesKey}.md`,
};
}
function listTasks(args: Record<string, unknown>, ctx: CallerContext) {
const status = statusFilter(args);
const rows = [];
for (const session of selectedSessions(args, ctx)) {
const sessionRows = withInbound(session, (db) =>
selectLiveTasks(db, status).map((row) => enrichListRow(db, toOutput(session, row))),
);
if (sessionRows) rows.push(...sessionRows);
}
return rows;
}
function getTask(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
for (const session of selectedSessions(args, ctx)) {
const found = withInbound(session, (db) => {
const row = selectTask(db, id);
if (!row) return undefined;
const seriesKey = row.series_id ?? row.row_id;
const stats = seriesStats(db, seriesKey);
const content = parseContent(row.content);
return {
...toOutput(session, row),
prompt: content.prompt,
script: content.script,
origin_session_id: content.originSessionId,
completed_runs: stats.runs,
failed_runs: stats.failed_runs,
recent_log: tailRunLog(session.agent_group_id, seriesKey),
};
});
if (found) return found;
}
throw new Error(`task not found: ${id}`);
}
function mutateTask(
args: Record<string, unknown>,
ctx: CallerContext,
fn: (db: Database.Database, id: string) => number,
) {
const id = taskId(args);
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, (db) => fn(db, id)) ?? 0;
}
if (touched === 0) throw new Error(`no live task matched: ${id}`);
return { series_id: id, touched };
}
function updateTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
const update: TaskUpdate = {};
if (typeof args.prompt === 'string') update.prompt = args.prompt;
if (args.process_after !== undefined) update.processAfter = parseProcessAfter(args.process_after);
const recurrence = normalizeNullableString(args.recurrence);
const script = normalizeNullableString(args.script);
if (recurrence !== undefined) {
validateRecurrence(recurrence);
// Effective script AFTER this update: the new value when provided
// (including an explicit clear), else whatever the task already has.
let scriptAfter: string | null = script !== undefined ? script : null;
if (script === undefined) {
for (const session of selectedSessions(args, ctx)) {
const row = withInbound(session, (db) => selectTask(db, id));
if (row) {
scriptAfter = parseContent(row.content).script;
break;
}
}
}
enforceRecurrenceLimit(recurrence, bool(args.dangerously_override_recurrence_limit), scriptAfter != null);
update.recurrence = recurrence;
}
if (script !== undefined) update.script = script;
const fields = Object.keys(update);
if (fields.length === 0) throw new Error('nothing to update');
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, (db) => updateTask(db, id, update)) ?? 0;
}
if (touched === 0) throw new Error(`no live task matched: ${id}`);
return { series_id: id, touched, fields };
}
function cancelTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
if (!bool(args.all)) {
return mutateTask(args, ctx, cancelTask);
}
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, cancelAllTasks) ?? 0;
}
return { cancelled: touched };
}
/**
* `ncl tasks run <id>` fire a task on demand without disturbing its schedule.
* Inserts a fresh pending occurrence (same series, content, no recurrence) due
* now, which the next sweep delivers through the normal fire path. Unlike
* `update --process-after now`, it neither consumes a one-shot nor force-advances
* a recurring series' armed occurrence, so it is safe for testing a task.
*/
function runTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
for (const session of selectedSessions(args, ctx)) {
const fired = withInbound(session, (db) => {
const row = selectTask(db, id);
if (!row) return undefined;
const seriesKey = row.series_id ?? row.row_id;
const rowId = makeTaskId(`${seriesKey}-run`);
// recurrence=NULL is load-bearing: a run-now row must not be re-armed by
// handleRecurrence into a phantom series.
insertTaskRow(db, {
id: rowId,
seriesId: seriesKey,
processAfter: new Date().toISOString(),
recurrence: null,
content: row.content,
});
return { series_id: seriesKey, row_id: rowId, status: 'pending' };
});
if (fired) return fired;
}
throw new Error(`task not found: ${id}`);
}
registerResource({
name: 'task',
plural: 'tasks',
table: 'messages_in',
description:
'Scheduled task — prompt plus run time. Tasks run from the agent group system session and the agent chooses delivery destination at fire time.',
idColumn: 'series_id',
scopeField: 'agent_group_id',
columns: [
{ name: 'series_id', type: 'string', description: 'Stable task handle.', generated: true },
{ name: 'agent_group_id', type: 'string', description: 'Agent group that owns the task.' },
{ name: 'session_id', type: 'string', description: 'System session that runs the task.' },
{ name: 'status', type: 'string', description: 'Live state.', enum: ['pending', 'paused'] },
{
name: 'process_after',
type: 'string',
// Not flagged required: with --recurrence the first run is derived from the
// cron grid (firstRunIso). Required only for one-shots, enforced in the
// create handler — so the generic col.required validator must stay off here.
description:
'Next run time (ISO 8601 or naive local). Required for one-shots; with --recurrence the first run is derived from the cron grid.',
updatable: true,
},
{ name: 'recurrence', type: 'string', description: 'Optional cron expression.', updatable: true },
{ name: 'prompt', type: 'string', description: 'Task prompt.', required: true, updatable: true },
{ name: 'script', type: 'string', description: 'Optional pre-task bash script.', updatable: true },
],
operations: {},
customOperations: {
list: {
access: 'open',
description: 'List live tasks with per-series run history (schedule, runs, failures, next fire).',
args: [
{ name: 'status', type: 'string', description: 'Filter by live state.', enum: ['pending', 'paused'] },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
{
name: 'all',
type: 'boolean',
description: 'List across all groups (host default when no --group; accepted for explicitness).',
},
],
handler: async (args, ctx) => listTasks(args, ctx),
// Server-rendered run-history table (frame `human` field) — the container
// agent gets the same legible view as the host CLI without a Bun-side
// formatter copy.
formatHuman: (rows) => formatTasksTable(rows as Parameters<typeof formatTasksTable>[0]),
},
get: {
access: 'open',
description: 'Get a task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => getTask(args, ctx),
},
create: {
access: 'open',
description:
`Create a scheduled task (recurring or one-shot) in the agent group system session.\n\n` +
`Requires --prompt plus EITHER --recurrence (recurring; first run derived from the cron grid) OR --process-after (one-shot, ISO 8601 or naive local). Always pass --name for a readable id.\n\n` +
`--script contract (pre-task gate, runs BEFORE the agent wakes):\n` +
` bash, 30s timeout, 1MB output cap. Its LAST stdout line must be JSON:\n` +
` {"wakeAgent": <bool>, "data": {...}}\n` +
` wakeAgent=false marks the run handled without waking the agent (zero tokens);\n` +
` wakeAgent=true wakes the agent with data attached to the prompt.\n` +
` DO: print the JSON as the very last line, exit 0, keep data small (a summary, not a dump).\n` +
` DON'T: print anything after the JSON, prompt for input, or rely on state from previous runs.\n` +
` Always test with bash -c '<script>' before scheduling.\n` +
` Persist state between fires under the group workspace (e.g. a last-seen id file).\n` +
` Use good judgement on whether to share with the user the script (only if they are technical), a description of the script condition, or whether there's no need.\n\n` +
`Frequency limit: recurrences more frequent than ${MAX_DAILY_FIRES} fires/day are refused unless the task\n` +
`carries a --script gate (the script decides whether each fire needs you — a gated fire that\n` +
`finds nothing costs zero tokens) or you pass --dangerously-override-recurrence-limit after\n` +
`the user explicitly confirmed they want an ungated frequent task.\n\n` +
`Failure backoff: a script that ERRORS repeatedly backs the series off (2,4,8,…60 min between fires; each errored fire counts as a failed run); after 8 consecutive failures the series is auto-paused with a note in its run log — fix the script, then \`ncl tasks resume <id>\`. A deliberate wakeAgent=false is a normal run and never backs off. \`ncl tasks get <id>\` shows failed_runs and the run log.`,
args: [
{
name: 'name',
type: 'string',
description: 'Short descriptive name → readable task id (<slug>-<hex>). Without it, ids are t-<hex>.',
},
{ name: 'prompt', type: 'string', description: 'Task prompt the agent wakes to.', required: true },
{
name: 'recurrence',
type: 'string',
description:
'Cron expression (instance TZ). First run derives from the cron grid when --process-after is omitted.',
},
{
name: 'dangerously_override_recurrence_limit',
type: 'boolean',
description:
'Schedule more than 4 fires/day anyway. Only after the user explicitly confirmed they understand the quota/token cost and you agree it is right.',
},
{
name: 'process_after',
type: 'string',
description: 'First/next run time (ISO 8601 or naive local). Required for one-shots.',
},
{
name: 'script',
type: 'string',
description: 'Pre-task gate script (bash) — see the --script contract above.',
},
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
],
examples: [
`# Recurring — --recurrence alone is enough; the first run comes off the cron grid:\nncl tasks create --name "sales briefing" --prompt "Send the weekday sales briefing" --recurrence "0 9 * * 1-5"`,
`# One-shot — --process-after required (UTC, offset, or naive-local in the instance TZ):\nncl tasks create --name "ping" --prompt "Remind me to call Dana" --process-after "tomorrow 18:00"`,
`# Monitor — script gates the run; the agent wakes only when something matters:\nncl tasks create --name "alert watch" --recurrence "*/15 * * * *" \\\n --prompt "Investigate the alerts in the script data and notify me if serious" \\\n --script 'c=$(curl -sf https://example.com/api/alerts | jq length) || exit 0\necho "{\\"wakeAgent\\": $([ "$c" -gt 0 ] && echo true || echo false), \\"data\\": {\\"alerts\\": $c}}"'`,
],
handler: async (args, ctx) => createTask(args, ctx),
},
'append-log': {
access: 'open',
description:
'Append a one-line run summary to a task run log (tasks/<id>.md).\n\nThe host stamps the UTC timestamp; you supply --msg. This is a LOG ENTRY, not a message — it sends nothing to anyone. Inside a task fire --id is auto-derived from your session. If you wrote or modified files during the run, name them in --msg.',
examples: [
`# Inside a task fire (--id auto-derived) — the run's work-log line:\nncl tasks append-log --msg "posted the daily digest to slack; one feed returned 403, skipped"`,
],
args: [
{
name: 'msg',
type: 'string',
description:
'Your work-log entry: what you did and why it mattered (like a human work log). The host prepends the UTC timestamp; this is logged, never sent to the user.',
required: true,
},
{
name: 'id',
type: 'string',
description: 'Task series id. Auto-derived when called from inside a task fire; required otherwise.',
},
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
],
handler: async (args, ctx) => appendTaskLog(args, ctx),
},
update: {
access: 'open',
description: 'Update a live task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{ name: 'prompt', type: 'string', description: 'Replace the task prompt.' },
{ name: 'process_after', type: 'string', description: 'New next-run time (ISO 8601 or naive local).' },
{ name: 'recurrence', type: 'string', description: 'New cron expression; "null"/"none" clears it (one-shot).' },
{
name: 'dangerously_override_recurrence_limit',
type: 'boolean',
description:
'Schedule more than 4 fires/day anyway. Only after the user explicitly confirmed they understand the quota/token cost and you agree it is right.',
},
{ name: 'script', type: 'string', description: 'New pre-task script; "null"/"none" removes it.' },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => updateTaskCommand(args, ctx),
},
cancel: {
access: 'open',
description: 'Cancel a live task by series id, or use --all as a kill switch.',
args: [
{ name: 'id', type: 'string', description: 'Task series id (omit with --all).' },
{ name: 'all', type: 'boolean', description: 'Cancel every live task in scope — kill switch.' },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => cancelTaskCommand(args, ctx),
},
run: {
access: 'open',
description:
'Fire a task now without changing its schedule (queues an extra run due immediately). Safe for testing — unlike update --process-after now, it neither consumes a one-shot nor advances a recurring series.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => runTaskCommand(args, ctx),
},
pause: {
access: 'open',
description: 'Pause a pending task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, pauseTask),
},
resume: {
access: 'open',
description: 'Resume a paused task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, resumeTask),
},
delete: {
access: 'open',
description: 'Hard-delete a task series and its history.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, deleteTask),
},
},
});
+64 -1
View File
@@ -10,7 +10,7 @@ import fs from 'fs';
import path from 'path';
import { describe, it, expect, afterEach } from 'vitest';
import { getInboundSourceSessionId, migrateMessagesInTable } from './session-db.js';
import { ensureSchema, getInboundSourceSessionId, migrateMessagesInTable, syncProcessingAcks } from './session-db.js';
const TEST_DIR = '/tmp/nanoclaw-session-db-test';
const DB_PATH = path.join(TEST_DIR, 'inbound.db');
@@ -92,3 +92,66 @@ describe('migrateMessagesInTable', () => {
db.close();
});
});
describe('syncProcessingAcks — script-skip counter', () => {
function freshPair() {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
fs.mkdirSync(TEST_DIR, { recursive: true });
ensureSchema(DB_PATH, 'inbound');
const outPath = path.join(TEST_DIR, 'outbound.db');
ensureSchema(outPath, 'outbound');
return { inDb: new Database(DB_PATH), outDb: new Database(outPath) };
}
function seedTask(inDb: InstanceType<typeof Database>, id: string, content: Record<string, unknown>) {
inDb
.prepare(
`INSERT INTO messages_in (id, seq, timestamp, status, tries, kind, content, series_id)
VALUES (?, 2, datetime('now'), 'processing', 0, 'task', ?, ?)`,
)
.run(id, JSON.stringify(content), id);
}
function ack(outDb: InstanceType<typeof Database>, id: string, status: string) {
outDb
.prepare(
"INSERT OR REPLACE INTO processing_ack (message_id, status, status_changed) VALUES (?, ?, datetime('now'))",
)
.run(id, status);
}
const status = (inDb: InstanceType<typeof Database>, id: string) =>
(inDb.prepare('SELECT status FROM messages_in WHERE id = ?').get(id) as { status: string }).status;
it('script-skip:error ack lands the row as a FAILED run (streak-derivable history)', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'script-skip:error');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('failed');
});
it('a settled row is terminal — a lingering ack cannot flip failed back to completed', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'script-skip:error');
syncProcessingAcks(inDb, outDb);
ack(outDb, 't1', 'completed');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('failed');
});
it('plain completed ack completes the row as before', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'completed');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('completed');
});
});
+20 -8
View File
@@ -168,15 +168,25 @@ export function getMessageForRetry(
export function syncProcessingAcks(inDb: Database.Database, outDb: Database.Database): void {
const completed = outDb
.prepare("SELECT message_id FROM processing_ack WHERE status IN ('completed', 'failed')")
.all() as Array<{ message_id: string }>;
.prepare(
"SELECT message_id, status FROM processing_ack WHERE status IN ('completed', 'failed', 'script-skip:error')",
)
.all() as Array<{ message_id: string; status: string }>;
if (completed.length === 0) return;
const updateStmt = inDb.prepare("UPDATE messages_in SET status = 'completed' WHERE id = ? AND status != 'completed'");
// `script-skip:error` (pre-task script crashed) lands as a FAILED run —
// semantically true, and it lets recurrence derive the trailing failed
// streak from the occurrence rows themselves (no stored counter).
const completeStmt = inDb.prepare(
"UPDATE messages_in SET status = 'completed' WHERE id = ? AND status NOT IN ('completed', 'failed')",
);
const failStmt = inDb.prepare(
"UPDATE messages_in SET status = 'failed' WHERE id = ? AND status NOT IN ('completed', 'failed')",
);
inDb.transaction(() => {
for (const { message_id } of completed) {
updateStmt.run(message_id);
for (const { message_id, status } of completed) {
(status === 'script-skip:error' ? failStmt : completeStmt).run(message_id);
}
})();
}
@@ -294,9 +304,11 @@ export function migrateDeliveredTable(db: Database.Database): void {
}
}
// Adds columns added to messages_in after the initial v2 schema to
// pre-existing session DBs. No-op on fresh installs where the columns are
// in the baseline schema. Backfills existing rows so invariants hold.
// LEGACY-COMPAT(v1-tasks): adds columns added to messages_in after the initial
// v2 schema to pre-existing session DBs — this lazy, on-open migration IS the
// upgrade path for old installs (there is no central migration for session
// DBs). No-op on fresh installs where the columns are in the baseline schema.
// Backfills existing rows so invariants hold (series_id = id).
export function migrateMessagesInTable(db: Database.Database): void {
const cols = new Set(
(db.prepare("PRAGMA table_info('messages_in')").all() as Array<{ name: string }>).map((c) => c.name),
+48 -1
View File
@@ -3,6 +3,8 @@ import { getDb, hasTable } from './connection.js';
// ── Sessions ──
export const TASKS_SYSTEM_THREAD_ID = 'system:tasks';
export function createSession(session: Session): void {
getDb()
.prepare(
@@ -55,7 +57,14 @@ export function findSessionForAgent(
/** Find an active session scoped to an agent group (ignoring messaging group). */
export function findSessionByAgentGroup(agentGroupId: string): Session | undefined {
return getDb()
.prepare("SELECT * FROM sessions WHERE agent_group_id = ? AND status = 'active' ORDER BY created_at DESC LIMIT 1")
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND status = 'active'
AND NOT (messaging_group_id IS NULL AND thread_id IS NOT NULL AND thread_id LIKE 'system:%')
ORDER BY created_at DESC
LIMIT 1`,
)
.get(agentGroupId) as Session | undefined;
}
@@ -63,6 +72,44 @@ export function getSessionsByAgentGroup(agentGroupId: string): Session[] {
return getDb().prepare('SELECT * FROM sessions WHERE agent_group_id = ?').all(agentGroupId) as Session[];
}
export function findSystemSession(agentGroupId: string, threadId: string): Session | undefined {
return getDb()
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND messaging_group_id IS NULL
AND thread_id = ?
AND status = 'active'
ORDER BY created_at DESC
LIMIT 1`,
)
.get(agentGroupId, threadId) as Session | undefined;
}
/** Per-task session thread id for a scheduled task series. */
export function taskThreadId(seriesId: string): string {
return `${TASKS_SYSTEM_THREAD_ID}:${seriesId}`;
}
/** True for any task session thread — a per-series one or the legacy shared one. */
export function isTaskThread(threadId: string | null): boolean {
return threadId === TASKS_SYSTEM_THREAD_ID || (threadId?.startsWith(`${TASKS_SYSTEM_THREAD_ID}:`) ?? false);
}
/** All active task sessions for a group — one per live series, plus any legacy shared one. */
export function findTaskSessions(agentGroupId: string): Session[] {
return getDb()
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND messaging_group_id IS NULL
AND status = 'active'
AND (thread_id = ? OR thread_id LIKE ?)
ORDER BY created_at DESC`,
)
.all(agentGroupId, TASKS_SYSTEM_THREAD_ID, `${TASKS_SYSTEM_THREAD_ID}:%`) as Session[];
}
export function getActiveSessions(): Session[] {
return getDb().prepare("SELECT * FROM sessions WHERE status = 'active'").all() as Session[];
}
+1 -1
View File
@@ -254,7 +254,7 @@ async function deliverMessage(
const content = JSON.parse(msg.content);
// System actions — handle internally (schedule_task, cancel_task, etc.)
// System actions — handle internally (cli_request, etc.)
if (msg.kind === 'system') {
await handleSystemAction(content, session, inDb);
return;
+20
View File
@@ -13,6 +13,7 @@ import {
_resetStuckProcessingRowsForTesting,
decideStuckAction,
parseSqliteUtc,
shouldCloseTaskSession,
} from './host-sweep.js';
import type { Session } from './types.js';
@@ -334,3 +335,22 @@ describe('parseSqliteUtc', () => {
expect(parseSqliteUtc(bare)).toBe(Date.parse(bare + 'Z'));
});
});
describe('shouldCloseTaskSession', () => {
it('closes a spent per-task session (no live tasks, no container)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', false, 0)).toBe(true);
});
it('keeps it while a task is still live (recurring re-armed, or pending/paused)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', false, 1)).toBe(false);
});
it('keeps it while its container is running (mid-fire)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', true, 0)).toBe(false);
});
it('never touches non-task sessions', () => {
expect(shouldCloseTaskSession('telegram:12345', false, 0)).toBe(false);
expect(shouldCloseTaskSession(null, false, 0)).toBe(false);
});
});
+28 -1
View File
@@ -30,7 +30,7 @@ import type Database from 'better-sqlite3';
import fs from 'fs';
import { ensureEgressNetwork } from './egress-lockdown.js';
import { getActiveSessions } from './db/sessions.js';
import { getActiveSessions, isTaskThread, updateSession } from './db/sessions.js';
import { getAgentGroup } from './db/agent-groups.js';
import {
countDueMessages,
@@ -167,6 +167,15 @@ async function sweep(): Promise<void> {
setTimeout(sweep, SWEEP_INTERVAL_MS);
}
/** A per-task session with no live tasks and no running container is spent → close it. */
export function shouldCloseTaskSession(
threadId: string | null,
containerRunning: boolean,
liveTaskCount: number,
): boolean {
return isTaskThread(threadId) && !containerRunning && liveTaskCount === 0;
}
async function sweepSession(session: Session): Promise<void> {
const agentGroup = getAgentGroup(session.agent_group_id);
if (!agentGroup) return;
@@ -234,6 +243,24 @@ async function sweepSession(session: Session): Promise<void> {
const { handleRecurrence } = await import('./modules/scheduling/recurrence.js');
await handleRecurrence(inDb, session);
// MODULE-HOOK:scheduling-recurrence:end
// 6. GC spent task sessions. An isolated per-task session with no live task
// rows left (one-shot fired, or all cancelled/deleted) and no container
// running is dead — close it so it stops being swept and listed. Runs after
// recurrence so a just-fired recurring series has already re-armed its next
// pending row and is never collected. The per-task log file in the workspace
// is the durable history and survives the close.
if (isTaskThread(session.thread_id)) {
const liveTasks = (
inDb
.prepare("SELECT COUNT(*) AS c FROM messages_in WHERE kind = 'task' AND status IN ('pending', 'paused')")
.get() as { c: number }
).c;
if (shouldCloseTaskSession(session.thread_id, isContainerRunning(session.id), liveTasks)) {
updateSession(session.id, { status: 'closed' });
log.info('Closed spent task session', { sessionId: session.id, threadId: session.thread_id });
}
}
} finally {
inDb.close();
outDb?.close();
-1
View File
@@ -18,7 +18,6 @@
// registers its handlers at import time.
import './approvals/index.js';
import './interactive/index.js';
import './scheduling/index.js';
import './permissions/index.js';
import './agent-to-agent/index.js';
import './self-mod/index.js';
-113
View File
@@ -1,113 +0,0 @@
/**
* Delivery action handlers for scheduling.
*
* The container can't write to inbound.db (host-owned). When the agent calls
* schedule_task / cancel_task / etc. via MCP, the container writes a
* `kind='system'` outbound message with an `action` field. The delivery path
* reaches into this module via the delivery-action registry and we apply the
* change to inbound.db here.
*/
import type Database from 'better-sqlite3';
import { wakeContainer } from '../../container-runner.js';
import { getSession } from '../../db/sessions.js';
import { log } from '../../log.js';
import { writeSessionMessage } from '../../session-manager.js';
import type { Session } from '../../types.js';
import { cancelTask, insertTask, pauseTask, resumeTask, updateTask, type TaskUpdate } from './db.js';
export async function handleScheduleTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
const prompt = content.prompt as string;
const script = content.script as string | null;
const processAfter = content.processAfter as string;
const recurrence = (content.recurrence as string) || null;
insertTask(inDb, {
id: taskId,
processAfter,
recurrence,
platformId: (content.platformId as string) ?? null,
channelType: (content.channelType as string) ?? null,
threadId: (content.threadId as string) ?? null,
content: JSON.stringify({ prompt, script }),
});
log.info('Scheduled task created', { taskId, processAfter, recurrence });
}
export async function handleCancelTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
cancelTask(inDb, taskId);
log.info('Task cancelled', { taskId });
}
export async function handlePauseTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
pauseTask(inDb, taskId);
log.info('Task paused', { taskId });
}
export async function handleResumeTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
resumeTask(inDb, taskId);
log.info('Task resumed', { taskId });
}
export async function handleUpdateTask(
content: Record<string, unknown>,
session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
const update: TaskUpdate = {};
if (typeof content.prompt === 'string') update.prompt = content.prompt;
if (typeof content.processAfter === 'string') update.processAfter = content.processAfter;
if (content.recurrence === null || typeof content.recurrence === 'string') {
update.recurrence = content.recurrence as string | null;
}
if (content.script === null || typeof content.script === 'string') {
update.script = content.script as string | null;
}
const touched = updateTask(inDb, taskId, update);
log.info('Task updated', { taskId, touched, fields: Object.keys(update) });
if (touched === 0) {
// Notify the agent that update_task matched nothing. Replicates the
// old notifyAgent helper that used to live in delivery.ts — inlined
// here so scheduling doesn't depend on delivery's private helpers.
writeSessionMessage(session.agent_group_id, session.id, {
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'chat',
timestamp: new Date().toISOString(),
platformId: session.agent_group_id,
channelType: 'agent',
threadId: null,
content: JSON.stringify({
text: `update_task: no live task matched id "${taskId}".`,
sender: 'system',
senderId: 'system',
}),
});
const fresh = getSession(session.id);
if (fresh) {
wakeContainer(fresh).catch((err) =>
log.error('Failed to wake container after update_task notification', { err }),
);
}
}
}
+37 -42
View File
@@ -10,7 +10,7 @@ import { describe, it, expect, afterEach } from 'vitest';
import { ensureSchema, openInboundDb } from '../../db/session-db.js';
import {
insertTask,
insertTaskRow,
insertRecurrence,
cancelTask,
pauseTask,
@@ -31,13 +31,11 @@ function freshDb() {
}
function insertBasicTask(db: ReturnType<typeof openInboundDb>, id: string, recurrence: string | null) {
insertTask(db, {
insertTaskRow(db, {
id,
seriesId: id,
processAfter: new Date().toISOString(),
recurrence,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'noop' }),
});
}
@@ -46,7 +44,7 @@ afterEach(() => {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
});
describe('insertTask', () => {
describe('insertTaskRow', () => {
it('stamps series_id = id on insert', () => {
const db = freshDb();
insertBasicTask(db, 'task-1', null);
@@ -68,13 +66,8 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: JSON.stringify({ prompt: 'noop' }),
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date(Date.now() + 86400000).toISOString());
@@ -93,7 +86,8 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
status: string;
recurrence: string | null;
};
expect(followUp.status).toBe('completed');
// Cancel marks 'cancelled' (not 'completed') so it never counts as a run.
expect(followUp.status).toBe('cancelled');
// Recurrence cleared so the sweep doesn't spawn another clone.
expect(followUp.recurrence).toBeNull();
db.close();
@@ -131,18 +125,37 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
expect(followUp.status).toBe('pending');
db.close();
});
it('pause/resume touch ONLY status — recurrence and process_after survive the cycle', () => {
const db = freshDb();
seedRecurringChain(db);
const before = db.prepare("SELECT recurrence, process_after FROM messages_in WHERE id = 'task-next'").get() as {
recurrence: string | null;
process_after: string | null;
};
pauseTask(db, 'task-next');
resumeTask(db, 'task-next');
const after = db.prepare("SELECT recurrence, process_after FROM messages_in WHERE id = 'task-next'").get() as {
recurrence: string | null;
process_after: string | null;
};
// A cancel-style copy-paste (clearing recurrence) would kill the series here.
expect(after.recurrence).toBe(before.recurrence);
expect(after.process_after).toBe(before.process_after);
db.close();
});
});
describe('updateTask', () => {
it('merges supplied fields into content JSON without clobbering others', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: new Date().toISOString(),
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'old', script: 'echo old', extra: 'keep me' }),
});
@@ -158,13 +171,11 @@ describe('updateTask', () => {
it('updates recurrence and process_after when supplied', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2026-01-01T00:00:00Z',
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
@@ -180,13 +191,11 @@ describe('updateTask', () => {
it('clears recurrence when null is passed', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2026-01-01T00:00:00Z',
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
@@ -200,26 +209,19 @@ describe('updateTask', () => {
it('reaches the live follow-up via series_id when called with the original id', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-orig',
seriesId: 'task-orig',
processAfter: new Date().toISOString(),
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'old' }),
});
db.prepare("UPDATE messages_in SET status = 'completed' WHERE id = 'task-orig'").run();
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: JSON.stringify({ prompt: 'old' }),
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date(Date.now() + 86400000).toISOString());
@@ -238,13 +240,11 @@ describe('updateTask', () => {
it('returns 0 when no live task matches', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: new Date().toISOString(),
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
db.prepare("UPDATE messages_in SET status = 'completed' WHERE id = 'task-1'").run();
@@ -262,13 +262,8 @@ describe('insertRecurrence', () => {
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: '{}',
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date().toISOString());
+88 -41
View File
@@ -14,43 +14,71 @@ import type Database from 'better-sqlite3';
import { nextEvenSeq } from '../../db/session-db.js';
export function insertTask(
/**
* Insert one pending task occurrence. `seriesId` is the series join key equal
* to `id` for a brand-new series, or the existing series for a recurrence clone
* or an on-demand run. Tasks never set platform/channel/thread (they fire into
* an isolated system session), so those columns are always NULL.
*/
export function insertTaskRow(
db: Database.Database,
task: {
row: {
id: string;
processAfter: string;
seriesId: string;
processAfter: string | null;
recurrence: string | null;
platformId: string | null;
channelType: string | null;
threadId: string | null;
content: string;
status?: 'pending' | 'paused';
},
): void {
db.prepare(
`INSERT INTO messages_in (id, seq, timestamp, status, tries, process_after, recurrence, kind, platform_id, channel_type, thread_id, content, series_id)
VALUES (@id, @seq, datetime('now'), 'pending', 0, @processAfter, @recurrence, 'task', @platformId, @channelType, @threadId, @content, @id)`,
VALUES (@id, @seq, datetime('now'), @status, 0, @processAfter, @recurrence, 'task', NULL, NULL, NULL, @content, @seriesId)`,
).run({
...task,
status: 'pending',
...row,
seq: nextEvenSeq(db),
});
}
export function cancelTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'completed', recurrence = NULL WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status IN ('pending', 'paused')",
).run(taskId, taskId);
// Cancel marks the live row 'cancelled' (not 'completed') so a never-fired
// occurrence is distinguishable from a real run and never inflates run history;
// recurrence is cleared so the series isn't re-armed by handleRecurrence.
export function cancelTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'cancelled', recurrence = NULL WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status IN ('pending', 'paused')",
)
.run(taskId, taskId).changes;
}
export function pauseTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'paused' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'pending'",
).run(taskId, taskId);
export function cancelAllTasks(db: Database.Database): number {
return db
.prepare(
"UPDATE messages_in SET status = 'cancelled', recurrence = NULL WHERE kind = 'task' AND status IN ('pending', 'paused')",
)
.run().changes;
}
export function resumeTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'pending' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'paused'",
).run(taskId, taskId);
export function pauseTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'paused' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'pending'",
)
.run(taskId, taskId).changes;
}
export function resumeTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'pending' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'paused'",
)
.run(taskId, taskId).changes;
}
export function deleteTask(db: Database.Database, taskId: string): number {
return db.prepare("DELETE FROM messages_in WHERE (id = ? OR series_id = ?) AND kind = 'task'").run(taskId, taskId)
.changes;
}
export interface TaskUpdate {
@@ -107,45 +135,64 @@ export function updateTask(db: Database.Database, taskId: string, update: TaskUp
return rows.length;
}
// Only tasks carry a recurrence (non-task writeSessionMessage never sets one),
// so getCompletedRecurring only ever returns task rows — the fields below are
// all that handleRecurrence needs to clone the next occurrence.
export interface RecurringMessage {
id: string;
kind: string;
content: string;
recurrence: string;
process_after: string | null;
platform_id: string | null;
channel_type: string | null;
thread_id: string | null;
series_id: string;
}
// Failed occurrences (script-skip:error runs) re-arm too — a broken monitor
// must keep its series alive so backoff can throttle it and the cap can pause
// it; dropping the row would silently kill the series on first script error.
export function getCompletedRecurring(db: Database.Database): RecurringMessage[] {
return db
.prepare("SELECT * FROM messages_in WHERE status = 'completed' AND recurrence IS NOT NULL")
.prepare("SELECT * FROM messages_in WHERE status IN ('completed', 'failed') AND recurrence IS NOT NULL")
.all() as RecurringMessage[];
}
/**
* Trailing consecutive FAILED occurrences of a series, newest backwards until
* the first completed run. This IS the script-failure streak derived from
* the occurrence history, no stored counter to update or reset. Deliberately
* counts ANY failed occurrence (script-skip:error acks AND stuck-message
* failures from host-sweep's MAX_TRIES path): a series failing for either
* reason should throttle, not spin.
*/
export function trailingFailedRuns(db: Database.Database, seriesKey: string): number {
const rows = db
.prepare(
`SELECT status FROM messages_in
WHERE (series_id = ? OR id = ?) AND kind = 'task' AND status IN ('completed', 'failed')
ORDER BY seq DESC`,
)
.all(seriesKey, seriesKey) as Array<{ status: string }>;
let streak = 0;
for (const r of rows) {
if (r.status !== 'failed') break;
streak++;
}
return streak;
}
export function insertRecurrence(
db: Database.Database,
msg: RecurringMessage,
newId: string,
nextRun: string | null,
status: 'pending' | 'paused' = 'pending',
): void {
db.prepare(
`INSERT INTO messages_in (id, seq, kind, timestamp, status, process_after, recurrence, platform_id, channel_type, thread_id, content, series_id)
VALUES (?, ?, ?, datetime('now'), 'pending', ?, ?, ?, ?, ?, ?, ?)`,
).run(
newId,
nextEvenSeq(db),
msg.kind,
nextRun,
msg.recurrence,
msg.platform_id,
msg.channel_type,
msg.thread_id,
msg.content,
msg.series_id,
);
insertTaskRow(db, {
id: newId,
seriesId: msg.series_id,
processAfter: nextRun,
recurrence: msg.recurrence,
content: msg.content,
status,
});
}
export function clearRecurrence(db: Database.Database, messageId: string): void {
-34
View File
@@ -1,34 +0,0 @@
/**
* Scheduling module one-shot and recurring tasks.
*
* Registers:
* - Five delivery action handlers: schedule_task, cancel_task, pause_task,
* resume_task, update_task. The container's scheduling MCP tools
* (container/agent-runner/src/mcp-tools/scheduling.ts) write system
* messages with these actions; the host applies them to inbound.db.
*
* Host integration points (filled by MODULE-HOOK markers, validated here
* with the scheduling module shipping inline):
* - `src/host-sweep.ts` MODULE-HOOK:scheduling-recurrence calls
* `handleRecurrence` each sweep tick.
* - `container/agent-runner/src/poll-loop.ts` MODULE-HOOK:scheduling-pre-task
* runs `applyPreTaskScripts` before the provider call so tasks carrying
* a pre-agent script can gate their own execution.
*
* No DB migration tasks are `messages_in` rows with `kind='task'`, so the
* module piggybacks on the core schema.
*/
import { registerDeliveryAction } from '../../delivery.js';
import {
handleCancelTask,
handlePauseTask,
handleResumeTask,
handleScheduleTask,
handleUpdateTask,
} from './actions.js';
registerDeliveryAction('schedule_task', handleScheduleTask);
registerDeliveryAction('cancel_task', handleCancelTask);
registerDeliveryAction('pause_task', handlePauseTask);
registerDeliveryAction('resume_task', handleResumeTask);
registerDeliveryAction('update_task', handleUpdateTask);
+106 -11
View File
@@ -8,13 +8,20 @@
*/
import fs from 'fs';
import path from 'path';
import { afterEach, describe, expect, it } from 'vitest';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { ensureSchema, openInboundDb } from '../../db/session-db.js';
import { insertTask } from './db.js';
import { handleRecurrence } from './recurrence.js';
import { insertTaskRow } from './db.js';
import { handleRecurrence, scriptBackoffMinutes } from './recurrence.js';
import type { Session } from '../../types.js';
// Pin a non-UTC zone so the tz-interpretation test is exact even on UTC CI.
// Asia/Tokyo is UTC+9 with no DST: "0 9 * * *" must land at 00:00:00Z sharp.
vi.mock('../../config.js', async (importOriginal) => {
const actual = await importOriginal<typeof import('../../config.js')>();
return { ...actual, TIMEZONE: 'Asia/Tokyo' };
});
const TEST_DIR = '/tmp/nanoclaw-recurrence-test';
const DB_PATH = path.join(TEST_DIR, 'inbound.db');
@@ -45,13 +52,11 @@ afterEach(() => {
describe('handleRecurrence', () => {
it('clones a completed recurring task with a next-run in the future', async () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: '0 9 * * *', // every day at 09:00 (user TZ)
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'daily digest' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-1'`).run();
@@ -77,15 +82,34 @@ describe('handleRecurrence', () => {
expect(new Date(follow.process_after).getTime()).toBeGreaterThan(Date.now());
});
it('interprets the cron expression in TIMEZONE, not UTC (the v1 regression)', async () => {
const db = freshDb();
insertTaskRow(db, {
id: 'task-tz',
seriesId: 'task-tz',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: '0 9 * * *', // 09:00 Asia/Tokyo === 00:00 UTC, exactly
content: JSON.stringify({ prompt: 'daily digest' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-tz'`).run();
await handleRecurrence(db, fakeSession());
const follow = db.prepare(`SELECT process_after FROM messages_in WHERE id != 'task-tz'`).get() as {
process_after: string;
};
// Drop the `{ tz: TIMEZONE }` option in recurrence.ts and this reads
// T09:00:00 (09:00 UTC) instead — red, even on a UTC CI runner.
expect(follow.process_after).toMatch(/T00:00:00/);
});
it('does not clone rows whose recurrence is already cleared', async () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'one-off' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-1'`).run();
@@ -96,3 +120,74 @@ describe('handleRecurrence', () => {
expect(count).toBe(1);
});
});
describe('handleRecurrence — script-failure backoff (streak derived from failed runs)', () => {
// A series whose last `fails` occurrences all landed as FAILED (script-skip:error
// runs, as synced by syncProcessingAcks). Only the newest row keeps recurrence —
// older occurrences had theirs cleared when they were re-armed. fails=0 seeds one
// healthy completed run.
function seedFailedStreak(db: ReturnType<typeof freshDb>, fails: number) {
const rows = Math.max(fails, 1);
for (let i = 0; i < rows; i++) {
insertTaskRow(db, {
id: `task-s-${i}`,
seriesId: 'task-s-0',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: i === rows - 1 ? '* * * * *' : null, // every minute — raw cron next is ~+1min
content: JSON.stringify({ prompt: 'monitor', script: 'exit 1' }),
});
db.prepare(`UPDATE messages_in SET status = ? WHERE id = ?`).run(
fails === 0 ? 'completed' : 'failed',
`task-s-${i}`,
);
}
return `task-s-${rows - 1}`; // the row carrying recurrence
}
const clone = (db: ReturnType<typeof freshDb>) =>
db.prepare(`SELECT status, process_after, recurrence FROM messages_in WHERE id NOT LIKE 'task-s-%'`).get() as {
status: string;
process_after: string;
recurrence: string | null;
};
it('exports the documented 2,4,8,…,60 progression', () => {
expect([1, 2, 3, 4, 5, 6, 7].map(scriptBackoffMinutes)).toEqual([2, 4, 8, 16, 32, 60, 60]);
});
it('pushes the clone past raw cron cadence while the script is failing', async () => {
const db = freshDb();
seedFailedStreak(db, 3); // streak 3 → backoff 8 min; cron next ≈ +1 min
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('pending');
const deltaMin = (new Date(next.process_after).getTime() - Date.now()) / 60_000;
expect(deltaMin).toBeGreaterThan(7); // backoff won over the 1-min cron grid
});
it('a healthy series (trailing run completed) re-arms on the raw cron grid', async () => {
const db = freshDb();
seedFailedStreak(db, 0);
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('pending');
const deltaMin = (new Date(next.process_after).getTime() - Date.now()) / 60_000;
expect(deltaMin).toBeLessThan(2); // no backoff applied
});
it('auto-pauses the series at the cap instead of re-arming', async () => {
const db = freshDb();
const liveId = seedFailedStreak(db, 8);
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('paused'); // `ncl tasks resume` revives in place
expect(next.recurrence).toBe('* * * * *');
const original = db.prepare(`SELECT recurrence FROM messages_in WHERE id = ?`).get(liveId) as {
recurrence: string | null;
};
expect(original.recurrence).toBeNull(); // not re-cloned next sweep
});
});
+60 -7
View File
@@ -11,27 +11,79 @@
* direct dynamic import. When scheduling moves to the modules branch in
* PR #8, the install skill re-fills the marker on install.
*/
import type Database from 'better-sqlite3';
import fs from 'fs';
import path from 'path';
import { TIMEZONE } from '../../config.js';
import type Database from 'better-sqlite3';
import { CronExpressionParser } from 'cron-parser';
import { GROUPS_DIR, TIMEZONE } from '../../config.js';
import { getAgentGroup } from '../../db/agent-groups.js';
import { log } from '../../log.js';
import type { Session } from '../../types.js';
import { clearRecurrence, getCompletedRecurring, insertRecurrence } from './db.js';
import { clearRecurrence, getCompletedRecurring, insertRecurrence, trailingFailedRuns } from './db.js';
// Consecutive pre-task-script failures (the series' trailing FAILED runs —
// derived from occurrence rows, no stored counter) throttle a broken monitor
// script instead of letting it wake a container at raw cron cadence forever.
// A deliberate wakeAgent=false gate is a normal completed run and never backs
// off. Mirrors the stuck-message retry in host-sweep.ts (BACKOFF_BASE_MS
// doubling, MAX_TRIES → failed): fail loud, don't spin.
const SCRIPT_FAIL_PAUSE_CAP = 8;
const SCRIPT_BACKOFF_CAP_MIN = 60;
/** 2, 4, 8, 16, 32, 60, 60… minutes for fails = 1, 2, 3… */
export function scriptBackoffMinutes(fails: number): number {
return Math.min(2 * 2 ** (fails - 1), SCRIPT_BACKOFF_CAP_MIN);
}
/** Host-written line in the series run log no agent session exists to call
* append-log when a script-gated series is auto-paused. Same format as
* appendTaskLog (tasks.ts). */
function appendHostTaskNote(agentGroupId: string, seriesId: string, note: string): void {
const ag = getAgentGroup(agentGroupId);
if (!ag) return;
const dir = path.join(GROUPS_DIR, ag.folder, 'tasks');
const timestamp = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
fs.mkdirSync(dir, { recursive: true });
fs.appendFileSync(path.join(dir, `${seriesId}.md`), `${timestamp}${note}\n`);
}
export async function handleRecurrence(inDb: Database.Database, session: Session): Promise<void> {
const recurring = getCompletedRecurring(inDb);
for (const msg of recurring) {
try {
const { CronExpressionParser } = await import('cron-parser');
// Interpret the cron expression in the user's timezone. v1 did this
// (src/v1/task-scheduler.ts:20-49); without it, a task written "0 9 * * *"
// by an agent running in a user's local TZ fires at 09:00 UTC instead of
// 09:00 user-local.
const interval = CronExpressionParser.parse(msg.recurrence, { tz: TIMEZONE });
const nextRun = interval.next().toISOString();
const prefix = msg.kind === 'task' ? 'task' : 'msg';
const newId = `${prefix}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const cronNext = interval.next().toDate();
const newId = `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const scriptFails = trailingFailedRuns(inDb, msg.series_id ?? msg.id);
if (scriptFails >= SCRIPT_FAIL_PAUSE_CAP) {
// Re-arm PAUSED at the cron time so `ncl tasks resume` revives the
// series in place; leave the why in the run log.
insertRecurrence(inDb, msg, newId, cronNext.toISOString(), 'paused');
clearRecurrence(inDb, msg.id);
appendHostTaskNote(
session.agent_group_id,
msg.series_id,
`auto-paused after ${scriptFails} consecutive script failures (host); fix the script, then \`ncl tasks resume ${msg.series_id}\``,
);
log.warn('Task series auto-paused: script keeps failing', {
seriesId: msg.series_id,
scriptFails,
sessionId: session.id,
});
continue;
}
const backoffAt = scriptFails > 0 ? Date.now() + scriptBackoffMinutes(scriptFails) * 60_000 : 0;
const nextRun = new Date(Math.max(cronNext.getTime(), backoffAt)).toISOString();
insertRecurrence(inDb, msg, newId, nextRun);
clearRecurrence(inDb, msg.id);
@@ -41,6 +93,7 @@ export async function handleRecurrence(inDb: Database.Database, session: Session
newId,
seriesId: msg.series_id,
nextRun,
...(scriptFails > 0 && { scriptFails, backoffMin: scriptBackoffMinutes(scriptFails) }),
sessionId: session.id,
});
} catch (err) {
+39
View File
@@ -22,9 +22,11 @@ import { ensureContainedInboxDir, isPathInside } from './inbox-safety.js';
import { getMessagingGroup } from './db/messaging-groups.js';
import {
createSession,
findSystemSession,
findSessionByAgentGroup,
findSessionForAgent,
getSession,
taskThreadId,
updateSession,
} from './db/sessions.js';
import {
@@ -120,6 +122,33 @@ export function resolveSession(
return { session, created: true };
}
/** Find or create the per-agent-group session used for scheduled tasks. */
/** Find or create the isolated session for one task series (thread `system:tasks:<seriesId>`). */
export function resolveTaskSession(agentGroupId: string, seriesId: string): { session: Session; created: boolean } {
const threadId = taskThreadId(seriesId);
const existing = findSystemSession(agentGroupId, threadId);
if (existing) return { session: existing, created: false };
const id = generateId();
const session: Session = {
id,
agent_group_id: agentGroupId,
messaging_group_id: null,
thread_id: threadId,
agent_provider: null,
status: 'active',
container_status: 'stopped',
last_active: null,
created_at: new Date().toISOString(),
};
createSession(session);
initSessionFolder(agentGroupId, id);
log.info('Task session created', { id, agentGroupId, seriesId });
return { session, created: true };
}
/** Create the session folder and initialize both DBs. */
export function initSessionFolder(agentGroupId: string, sessionId: string): void {
const dir = sessionDir(agentGroupId, sessionId);
@@ -350,6 +379,16 @@ export function openInboundDb(agentGroupId: string, sessionId: string): Database
return db;
}
/** Open a session's inbound DB, run `fn`, and always close it. */
export function withInboundDb<T>(agentGroupId: string, sessionId: string, fn: (db: Database.Database) => T): T {
const db = openInboundDb(agentGroupId, sessionId);
try {
return fn(db);
} finally {
db.close();
}
}
/** Open the outbound DB for a session (host reads only). */
export function openOutboundDb(agentGroupId: string, sessionId: string): Database.Database {
return openOutboundDbRaw(outboundDbPath(agentGroupId, sessionId));
+28 -1
View File
@@ -1,6 +1,6 @@
import { describe, it, expect } from 'vitest';
import { formatLocalTime, isValidTimezone, resolveTimezone } from './timezone.js';
import { formatLocalTime, isValidTimezone, parseZonedToUtc, resolveTimezone } from './timezone.js';
// --- formatLocalTime ---
@@ -62,3 +62,30 @@ describe('resolveTimezone', () => {
expect(resolveTimezone('')).toBe('UTC');
});
});
describe('parseZonedToUtc', () => {
const iso = (s: string, tz: string): string => parseZonedToUtc(s, tz).toISOString();
it('reads a naive timestamp as wall-clock in a fixed-offset zone', () => {
expect(iso('2026-06-20T09:00:00', 'Asia/Tokyo')).toBe('2026-06-20T00:00:00.000Z'); // UTC+9
});
it('applies the correct seasonal offset (DST honored)', () => {
// Same wall-clock time, different UTC offset by season — proves the offset
// is computed against the zone's rules, not a fixed guess.
expect(iso('2026-07-01T12:00:00', 'America/New_York')).toBe('2026-07-01T16:00:00.000Z'); // EDT -4
expect(iso('2026-01-01T12:00:00', 'America/New_York')).toBe('2026-01-01T17:00:00.000Z'); // EST -5
});
it('passes a trailing-Z timestamp through unchanged', () => {
expect(iso('2026-06-20T09:00:00Z', 'Asia/Tokyo')).toBe('2026-06-20T09:00:00.000Z');
});
it('passes an explicit offset through', () => {
expect(iso('2026-06-20T09:00:00+02:00', 'Asia/Tokyo')).toBe('2026-06-20T07:00:00.000Z');
});
it('falls back to UTC for an invalid zone', () => {
expect(iso('2026-06-20T09:00:00', 'Not/AZone')).toBe('2026-06-20T09:00:00.000Z');
});
});
+42
View File
@@ -35,3 +35,45 @@ export function formatLocalTime(utcIso: string, timezone: string): string {
hour12: true,
});
}
/**
* Interpret a naive ISO-like timestamp (no trailing `Z`, no offset) as wall-clock
* time in `tz` and return the corresponding UTC Date. Strings that already carry
* offset info (`Z` or `+-HH:MM`) are passed through to the Date constructor.
*/
export function parseZonedToUtc(input: string, tz: string): Date {
const hasOffset = /Z$|[+-]\d{2}:?\d{2}$/.test(input.trim());
if (hasOffset) return new Date(input);
const zone = resolveTimezone(tz);
const asIfUtc = new Date(input + 'Z');
if (Number.isNaN(asIfUtc.getTime())) return asIfUtc;
const fmt = new Intl.DateTimeFormat('en-US', {
timeZone: zone,
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hour12: false,
});
const parts = Object.fromEntries(
fmt
.formatToParts(asIfUtc)
.filter((p) => p.type !== 'literal')
.map((p) => [p.type, p.value]),
);
const hour = parts.hour === '24' ? '00' : parts.hour;
const zonedAsUtcMs = Date.UTC(
Number(parts.year),
Number(parts.month) - 1,
Number(parts.day),
Number(hour),
Number(parts.minute),
Number(parts.second),
);
const offsetMs = zonedAsUtcMs - asIfUtc.getTime();
return new Date(asIfUtc.getTime() - offsetMs);
}
+1 -1
View File
@@ -142,7 +142,7 @@ export interface Session {
// ── Session DB entities ──
export type MessageInKind = 'chat' | 'chat-sdk' | 'task' | 'webhook' | 'system';
export type MessageInStatus = 'pending' | 'processing' | 'completed' | 'failed';
export type MessageInStatus = 'pending' | 'processing' | 'completed' | 'failed' | 'cancelled';
export interface MessageIn {
id: string;