mirror of
https://github.com/qwibitai/nanoclaw.git
synced 2026-07-09 18:57:08 +08:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5459925c65 | |||
| b0c76ce4c5 | |||
| 87a2d3af8d | |||
| 68ebc02114 | |||
| d4e73c5523 | |||
| 9a103f4f93 | |||
| a8554c6248 | |||
| 2480ae7cb8 | |||
| 8943c1cbf4 |
@@ -14,6 +14,7 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
pull-requests: write
|
||||
issues: write # createLabel — auto-provisions the core-team label
|
||||
steps:
|
||||
- uses: actions/github-script@v7
|
||||
with:
|
||||
@@ -30,6 +31,24 @@ jobs:
|
||||
|
||||
if (body.includes('contributing-guide: v1')) labels.push('follows-guidelines');
|
||||
|
||||
// Lowercase GitHub logins; keep in sync with the core team roster.
|
||||
const CORE_TEAM = ['gavrielc', 'koshkoshinsk', 'glifocat', 'gabi-simons', 'omri-maya', 'amit-shafnir', 'moshe-nanoco'];
|
||||
const author = context.payload.pull_request.user.login.toLowerCase();
|
||||
if (CORE_TEAM.includes(author)) {
|
||||
labels.push('core-team');
|
||||
try {
|
||||
await github.rest.issues.createLabel({
|
||||
owner: context.repo.owner,
|
||||
repo: context.repo.repo,
|
||||
name: 'core-team',
|
||||
color: '1D76DB',
|
||||
description: 'PR opened by a core team member',
|
||||
});
|
||||
} catch (e) {
|
||||
if (e.status !== 422) throw e; // 422: label already exists
|
||||
}
|
||||
}
|
||||
|
||||
if (labels.length > 0) {
|
||||
await github.rest.issues.addLabels({
|
||||
owner: context.repo.owner,
|
||||
|
||||
@@ -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^(n−1) 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.
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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(' · ')})` : '';
|
||||
}
|
||||
|
||||
@@ -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]);
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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
@@ -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 |
|
||||
|
||||
---
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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")]
|
||||
|
||||
@@ -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):
|
||||
|
||||
|
||||
@@ -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.
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "nanoclaw",
|
||||
"version": "2.1.39",
|
||||
"version": "2.1.40",
|
||||
"description": "Personal Claude assistant. Lightweight, secure, customizable.",
|
||||
"type": "module",
|
||||
"packageManager": "pnpm@10.33.0",
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="90" height="20" role="img" aria-label="209k tokens, 104% of context window">
|
||||
<title>209k tokens, 104% of context window</title>
|
||||
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="90" height="20" role="img" aria-label="213k tokens, 106% of context window">
|
||||
<title>213k tokens, 106% of context window</title>
|
||||
<linearGradient id="s" x2="0" y2="100%">
|
||||
<stop offset="0" stop-color="#bbb" stop-opacity=".1"/>
|
||||
<stop offset="1" stop-opacity=".1"/>
|
||||
@@ -15,8 +15,8 @@
|
||||
<g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" font-size="11">
|
||||
<text aria-hidden="true" x="26" y="15" fill="#010101" fill-opacity=".3">tokens</text>
|
||||
<text x="26" y="14">tokens</text>
|
||||
<text aria-hidden="true" x="71" y="15" fill="#010101" fill-opacity=".3">209k</text>
|
||||
<text x="71" y="14">209k</text>
|
||||
<text aria-hidden="true" x="71" y="15" fill="#010101" fill-opacity=".3">213k</text>
|
||||
<text x="71" y="14">213k</text>
|
||||
</g>
|
||||
</g>
|
||||
</a>
|
||||
|
||||
|
Before Width: | Height: | Size: 1.1 KiB After Width: | Height: | Size: 1.1 KiB |
@@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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}`,
|
||||
|
||||
@@ -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' });
|
||||
|
||||
|
||||
@@ -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
@@ -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';
|
||||
|
||||
|
||||
@@ -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.',
|
||||
|
||||
@@ -14,3 +14,4 @@ import './user-dms.js';
|
||||
import './dropped-messages.js';
|
||||
import './approvals.js';
|
||||
import './sessions.js';
|
||||
import './tasks.js';
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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),
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -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
@@ -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
@@ -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
@@ -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;
|
||||
|
||||
@@ -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
@@ -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();
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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 }),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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);
|
||||
@@ -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
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
@@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user