Compare commits

..

2 Commits

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

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

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

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

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

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-08 17:04:45 +03:00
70 changed files with 2839 additions and 1802 deletions
+2 -1
View File
@@ -4,7 +4,8 @@ All notable changes to NanoClaw will be documented in this file.
## [Unreleased]
- **One approval contract across every hold.** All approval holds now share one hold-record shape on `pending_approvals` (approver rule, approver blast-radius scope, dedup key — migration 019, with backfills) and ONE click-authorization rule (`mayResolve` in `src/modules/approvals/approver-rule.ts`), replacing three divergent click-auth copies. Unknown-sender admission folds onto the approvals primitive (sessionless hold, action `sender_admit`) — the `pending_sender_approvals` table is dropped by migration; an in-flight sender card does not survive the upgrade (a new message from the same sender re-triggers one). Channel registration and OneCLI keep their flows and adopt the shared rule. Observers now see the full hold lifecycle with zero touch points inside the flows: `registerApprovalRequestedHandler` (new, the creation-side sibling of `registerApprovalResolvedHandler`) fires once whenever a hold record comes into existence, whichever stack created it — `requestApproval`, the OneCLI credential bridge, channel registration (as a synthesized hold view) — and every resolution announces through the approval-resolved observer (outcome `approve` | `reject` | `expire` | `sweep`, session nullable). Two absorbed security fixes intentionally change decision outcomes: **(D1)** a hold with global blast radius (e.g. `roles grant`) can only be approved by an owner or global admin — a scoped admin's click is now rejected; **(D4, approver half)** channel-registration approvers are owners/global admins, no longer "admin of whichever agent group sorts first".
- **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. Also fixed: an explicitly-addressed `<message to>` in a task fire's final text now delivers as a deliberate send (previously suppressed as a turn-reply echo → zero delivery when the agent skipped the MCP tool); identical echoes of an MCP send are dropped in the runner, where the duplication originates.
- [BREAKING] **Scheduled tasks moved from MCP tools to `ncl tasks`.** The six scheduling MCP tools are no longer exposed to agent containers; agents and operators manage tasks with `ncl tasks list/get/create/update/cancel/pause/resume/delete`. New tasks run from a per-agent-group system session rather than waking the chat session that created them, and task writes are not approval-gated inside the owning group. **Migration:** [docs/ncl-tasks-migration.md](docs/ncl-tasks-migration.md).
- **Optional per-container resource caps.** `CONTAINER_CPU_LIMIT` and `CONTAINER_MEMORY_LIMIT` pass through to `docker run` as `--cpus` / `--memory` (`container-runner.ts`). Both empty by default — no flag added, spawn args byte-identical to today — so existing installs are unaffected. Set them to cap an agent container's CPU/memory so one agent can't monopolize the host (e.g. `CONTAINER_CPU_LIMIT=2`, `CONTAINER_MEMORY_LIMIT=8g`). Swap is intentionally not managed here: `--memory` is a hard cap on a swapless host.
- [BREAKING] **Chat SDK pinned to `4.29.0` (was `4.26.0` via `^4.24.0`).** `chat` and the `@chat-adapter/*` channel adapters are version-locked — the adapter's `ChatInstance` must match the bridge's, so a mismatched pair fails to typecheck at `createChatSdkBridge(...)`. `chat` is therefore pinned exactly, and the channel-adapter install pins move with it — the `/add-<channel>` SKILL.md steps and `setup/*.sh` scripts on `main`, plus the adapter code on the `channels` branch. Core installs with no channel (only `cli`) are unaffected. **Migration:** if any channel is installed (Slack, Discord, Telegram, Teams, …), re-run its `/add-<channel>` skill to pull the matching `4.29.0` adapter.
- **Budget/billing-exhausted LLM turns now reach the user instead of being silently dropped.** When a turn ends in a non-retryable provider error (e.g. an Anthropic `403 billing_error`) with no `<message>` wrapping, the agent-runner delivers the provider's notice to the originating channel and stops re-nudging the failing gateway. `providers/claude.ts` now surfaces the SDK's `is_error` flag (and the error subtype's `errors[]` text); `poll-loop.ts` delivers that text and skips the re-wrap retry. Fixes the case where a spend-limit notice produced silence plus a turn-after-turn retry loop.
+3 -3
View File
@@ -67,8 +67,7 @@ For ad-hoc queries from skills or scripts, use the in-tree wrapper rather than t
| `src/container-runner.ts` | Spawns per-agent-group Docker containers with session DB + outbox mounts, OneCLI `ensureAgent` |
| `src/container-runtime.ts` | Docker CLI wrapper (runtime binary, host-gateway args, mount args), orphan cleanup |
| `src/modules/permissions/access.ts` | `canAccessAgentGroup` — owner / global admin / scoped admin / member resolution against `user_roles` + `agent_group_members` |
| `src/modules/approvals/primitive.ts` | `pickApprover`, `pickApprovalDelivery`, `requestApproval`, approval-handler registry, hold-lifecycle observers (`registerApprovalRequestedHandler` / `registerApprovalResolvedHandler` — every stack announces creation + resolution through these) |
| `src/modules/approvals/approver-rule.ts` | `mayResolve` — the one click-authorization rule (approver rule `exclusive` \| `admins-of-scope` + blast-radius scope) for every hold |
| `src/modules/approvals/primitive.ts` | `pickApprover`, `pickApprovalDelivery`, `requestApproval`, approval-handler registry |
| `src/command-gate.ts` | Router-side admin command gate — queries `user_roles` directly (no env var, no container-side check) |
| `src/modules/approvals/onecli-approvals.ts` | OneCLI credentialed-action approval bridge |
| `src/modules/permissions/user-dm.ts` | Cold-DM resolution + `user_dms` cache |
@@ -106,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) |
@@ -138,7 +138,7 @@ Per-agent-group container runtime config (provider, model, packages, MCP servers
| Value | Behavior |
|-------|----------|
| `disabled` | Agent never learns about ncl (instructions excluded from CLAUDE.md). Host dispatch rejects any `cli_request`. |
| `group` (default) | Agent can access `groups`, `sessions`, `destinations`, `members` only, scoped to its own agent group. `--id` and group args are auto-filled. Cross-group access rejected. `cli_scope` changes blocked. |
| `group` (default) | Agent can access `groups`, `sessions`, `destinations`, `members`, `tasks` only, scoped to its own agent group. `--id` and group args are auto-filled. Cross-group access rejected. `cli_scope` changes blocked. |
| `global` | Unrestricted. Set automatically for owner agent groups via `init-first-agent`. |
Key files: `src/db/container-configs.ts`, `src/container-config.ts`, `src/cli/dispatch.ts` (scope enforcement), `src/claude-md-compose.ts` (instructions exclusion).
@@ -120,6 +120,24 @@ export function markCompleted(ids: string[]): void {
})();
}
/**
* Ack task messages whose pre-task script gated the run. The reason decides
* the ack: `gated` (wakeAgent=false) is the monitor working as designed → a
* plain `completed`; `error` (broken script) → `script-skip:error`, which the
* host's ack sync records as a FAILED run so recurrence can read the trailing
* failed streak off the occurrence rows and back the series off.
*/
export function markScriptSkipped(skips: Array<{ id: string; reason: string }>): void {
if (skips.length === 0) return;
const db = getOutboundDb();
const stmt = db.prepare(
"INSERT OR REPLACE INTO processing_ack (message_id, status, status_changed) VALUES (?, ?, datetime('now'))",
);
db.transaction(() => {
for (const s of skips) stmt.run(s.id, s.reason === 'error' ? 'script-skip:error' : 'completed');
})();
}
/** Mark a single message as failed — writes to processing_ack in outbound.db. */
export function markFailed(id: string): void {
getOutboundDb()
@@ -141,3 +141,22 @@ export function getUndeliveredMessages(): MessageOutRow[] {
)
.all() as MessageOutRow[];
}
/**
* True if a deliberate send with this exact destination + text already exists
* (an MCP send_message row from the current turn). Used by the task-fire
* final-text dispatcher to drop the turn-final <message> echo of a send the
* agent already made — the dedup happens where the duplication originates.
*/
export function hasIdenticalSend(platformId: string, channelType: string, text: string): boolean {
const row = getOutboundDb()
.prepare(
`SELECT 1 FROM messages_out
WHERE platform_id = $platform_id AND channel_type = $channel_type
AND (in_reply_to IS NULL OR in_reply_to = '')
AND json_extract(content, '$.text') = $text
LIMIT 1`,
)
.get({ $platform_id: platformId, $channel_type: channelType, $text: text });
return row != null;
}
+9 -4
View File
@@ -105,13 +105,11 @@ function buildDestinationsSection(): string {
const lines = ['## Sending messages', ''];
if (all.length === 1) {
const d = all[0];
const label = d.displayName && d.displayName !== d.name ? ` (${d.displayName})` : '';
lines.push(`Your destination is \`${d.name}\`${label}.`);
lines.push(`Your destination is \`${d.name}\`${destinationLabel(d)}.`);
} else {
lines.push('You can send messages to the following destinations:', '');
for (const d of all) {
const label = d.displayName && d.displayName !== d.name ? ` (${d.displayName})` : '';
lines.push(`- \`${d.name}\`${label}`);
lines.push(`- \`${d.name}\`${destinationLabel(d)}`);
}
}
lines.push('');
@@ -128,3 +126,10 @@ function buildDestinationsSection(): string {
);
return lines.join('\n');
}
function destinationLabel(d: DestinationEntry): string {
const parts: string[] = [];
if (d.channelType) parts.push(d.channelType);
if (d.displayName && d.displayName !== d.name) parts.push(d.displayName);
return parts.length > 0 ? ` (${parts.join(' · ')})` : '';
}
+6
View File
@@ -98,6 +98,11 @@ export interface RoutingContext {
channelType: string | null;
threadId: string | null;
inReplyTo: string | null;
/** Batch is a task fire — explicit `<message to>` sends must NOT inherit
* inReplyTo (the series id), or the host's task-fire suppression drops
* them as turn-final echoes: zero delivery. Deliberate sends are
* in_reply_to-null, same as the out-of-process MCP send_message path. */
taskFire: boolean;
}
/**
@@ -111,6 +116,7 @@ export function extractRouting(messages: MessageInRow[]): RoutingContext {
channelType: first?.channel_type ?? null,
threadId: first?.thread_id ?? null,
inReplyTo: first?.id ?? null,
taskFire: messages.length > 0 && messages.every((m) => m.kind === 'task'),
};
}
@@ -18,12 +18,13 @@ Your CLI access may be scoped. Run `ncl help` to see which resources are availab
Run `ncl help` for the full list. Common resources:
| Resource | Verbs | What it is |
|----------|-------|------------|
| groups | list, get, create, update, delete, restart, config get/update, config add-mcp-server/remove-mcp-server, config add-package/remove-package | Agent groups (workspace, personality, container config) |
| sessions | list, get | Active sessions (read-only) |
| destinations | list, add, remove | Where an agent group can send messages |
| members | list, add, remove | Unprivileged access gate for an agent group |
| Resource | Verbs | What it is |
| ------------ | ----------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------- |
| groups | list, get, create, update, delete, restart, config get/update, config add-mcp-server/remove-mcp-server, config add-package/remove-package | Agent groups (workspace, personality, container config) |
| sessions | list, get | Active sessions (read-only) |
| destinations | list, add, remove | Where an agent group can send messages |
| members | list, add, remove | Unprivileged access gate for an agent group |
| tasks | list, get, create, update, cancel, pause, resume, delete, append-log | Scheduled tasks for your agent group |
Additional resources (available under `global` scope only): messaging-groups, wirings, users, roles, user-dms, dropped-messages, approvals.
@@ -33,11 +34,12 @@ Additional resources (available under `global` scope only): messaging-groups, wi
- **Restarting your container** — `ncl groups restart` (with optional `--rebuild` and `--message`).
- **Checking who's in your group** — `ncl members list`.
- **Seeing your destinations** — `ncl destinations list`.
- **Scheduling work** — `ncl tasks create`, then `ncl tasks list/get/update/cancel/pause/resume/delete`; `ncl tasks run <id>` fires one extra run now (testing) without changing the schedule. At the end of each task run, `ncl tasks append-log --msg "…"` to record what happened (host-timestamped, not a message).
- **Answering questions about the system** — query `ncl` rather than guessing.
### Access rules
Read commands (list, get) are open. Write commands (create, update, delete, restart, config update, add, remove) require admin approval — the request is held until an admin approves it.
Read commands (list, get) are open. Most write commands (create, update, delete, restart, config update, add, remove) require admin approval — the request is held until an admin approves it. `ncl tasks` is the exception: an agent can manage its own group tasks without approval.
### Approval flow
@@ -61,6 +63,13 @@ ncl groups config get
ncl sessions list
ncl destinations list
ncl members list
ncl tasks list
# Always pass a short descriptive --name so the task id is readable (e.g. daily-briefing-a25c, not a long uuid).
# For a recurring task, --recurrence alone sets the schedule (first run derived from it); add --process-after only for one-shots.
ncl tasks create --name "daily briefing" --prompt "Send the daily briefing" --recurrence "0 9 * * *"
# At the END of every task run, record one line of history. The host stamps the UTC time — you supply only the summary.
# This is a LOG ENTRY, not a message: it sends nothing to anyone. Inside a task fire --id is auto-derived from your session.
ncl tasks append-log --msg "posted the daily digest to slack; one feed returned 403, skipped"
# Write commands (approval required)
ncl groups restart
@@ -6,7 +6,6 @@
* at module scope, and append the import here. No central list.
*/
import './core.js';
import './scheduling.js';
import './interactive.js';
import './agents.js';
import './self-mod.js';
@@ -1,40 +1,25 @@
## Task scheduling (`schedule_task`)
## Task scheduling (`ncl tasks`)
For any recurring task, use `schedule_task`. This is the scheduling path — tasks persist across sessions and restarts, and support the pre-task `script` hook described below.
Use `ncl tasks` for one-shot and recurring tasks. A task runs in this agent group's system session, not in the current chat session, so when it fires you must choose a destination explicitly with `<message to="name">...</message>` or `send_message({ to: "name", ... })`.
To inspect or change existing tasks, use `list_tasks` (returns one row per series with the stable id) and `update_task` / `cancel_task` / `pause_task` / `resume_task`. Prefer `update_task` over cancel + reschedule.
Pass `--name "<short label>"` on create to get a readable task id (e.g. `--name "sales briefing"` `sales-briefing-a25c`); without it ids are `t-<hex>`.
Frequent recurring scheduled tasks — more than a few times a day — consume API credits and can risk account restrictions. You can add a `script` that runs first, and you will only be called when the check passes.
### How it works
1. Provide a bash `script` alongside the `prompt` when scheduling
2. When the task fires, the script runs first
3. Script returns: `{ "wakeAgent": true/false, "data": {...} }`
4. If `wakeAgent: false` — nothing happens, task waits for next run
5. If `wakeAgent: true` — claude receives the script's data + prompt and handles
### Always test your script first
Before scheduling, run the script directly to verify it works:
Common commands:
```bash
bash -c 'node --input-type=module -e "
const r = await fetch(\"https://api.github.com/repos/owner/repo/pulls?state=open\");
const prs = await r.json();
console.log(JSON.stringify({ wakeAgent: prs.length > 0, data: prs.slice(0, 5) }));
"'
ncl tasks create --name "ping" --prompt "Remind me to call Dana" --process-after "tomorrow 18:00"
ncl tasks list
ncl tasks get ping-a25c # includes run count, failures, and recent run-log lines
ncl tasks run ping-a25c # fire once now without changing the schedule (testing)
ncl tasks update ping-a25c --prompt "New instructions"
ncl tasks pause ping-a25c
ncl tasks resume ping-a25c
ncl tasks cancel ping-a25c # or --all as a kill switch
ncl tasks delete ping-a25c
```
### When NOT to use scripts
Use good judgement on whether it's appropriate to check in with the user about the task prompt before task creation, and if so, whether to share verbatim or a description of it.
If a task requires your judgment every time (daily briefings, reminders, reports), skip the script — just use a regular prompt. Do not attempt to do things like sentiment analysis or advanced nlp in scripts.
`--process-after` accepts UTC timestamps or naive local timestamps interpreted in the instance timezone (shown in the `<context timezone="..."/>` header).
### Frequent task guidance
If a user wants a task to run more than a few times a day and a script can't be used:
- Explain that each time the task fires it uses API credits and risks rate limits
- Suggest adjusting the task requirements in a way that will allow you to use a script
- If the user needs an LLM to evaluate data, suggest using an API key with direct Anthropic API calls inside the script
- Help the user find the minimum viable frequency
Run `ncl tasks create --help` for schedules, options, and pre-task gate scripts (checks that run before you wake).
@@ -1,302 +0,0 @@
/**
* Scheduling MCP tools: schedule_task, list_tasks, cancel_task, pause_task, resume_task.
*
* With the two-DB split, the container cannot write to inbound.db (host-owned).
* Scheduling operations are sent as system actions via messages_out — the host
* reads them during delivery and applies the changes to inbound.db.
*/
import { getInboundDb } from '../db/connection.js';
import { writeMessageOut } from '../db/messages-out.js';
import { getSessionRouting } from '../db/session-routing.js';
import { TIMEZONE, parseZonedToUtc } from '../timezone.js';
import { registerTools } from './server.js';
import type { McpToolDefinition } from './types.js';
function log(msg: string): void {
console.error(`[mcp-tools] ${msg}`);
}
function generateId(): string {
return `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
}
function routing() {
return getSessionRouting();
}
function ok(text: string) {
return { content: [{ type: 'text' as const, text }] };
}
function err(text: string) {
return { content: [{ type: 'text' as const, text: `Error: ${text}` }], isError: true };
}
export const scheduleTask: McpToolDefinition = {
tool: {
name: 'schedule_task',
description:
`Schedule a one-shot or recurring task. The user's timezone is declared in the <context timezone="..."/> header of your prompt — interpret the user's "9pm" etc. in that zone. Cron expressions are interpreted in the user's timezone too.`,
inputSchema: {
type: 'object' as const,
properties: {
prompt: { type: 'string', description: 'Task instructions/prompt' },
processAfter: {
type: 'string',
description:
`ISO 8601 timestamp for the first run. Accepts either UTC (ending in "Z" or "+00:00") or a naive local timestamp (no offset) which is interpreted in the user's timezone (e.g. "2026-01-15T21:00:00" = 9pm user-local). Prefer naive local.`,
},
recurrence: {
type: 'string',
description:
'Cron expression for recurring tasks (e.g., "0 9 * * 1-5" = weekdays at 9am user-local). Evaluated in the user\'s timezone.',
},
script: { type: 'string', description: 'Optional pre-agent script to run before processing' },
},
required: ['prompt', 'processAfter'],
},
},
async handler(args) {
const prompt = args.prompt as string;
const processAfterIn = args.processAfter as string;
if (!prompt || !processAfterIn) return err('prompt and processAfter are required');
let processAfter: string;
try {
const d = parseZonedToUtc(processAfterIn, TIMEZONE);
if (Number.isNaN(d.getTime())) return err(`invalid processAfter: ${processAfterIn}`);
processAfter = d.toISOString();
} catch {
return err(`invalid processAfter: ${processAfterIn}`);
}
const id = generateId();
const r = routing();
const recurrence = (args.recurrence as string) || null;
const script = (args.script as string) || null;
// Write as a system action — host will insert into inbound.db
writeMessageOut({
id,
kind: 'system',
platform_id: r.platform_id,
channel_type: r.channel_type,
thread_id: r.thread_id,
content: JSON.stringify({
action: 'schedule_task',
taskId: id,
prompt,
script,
processAfter,
recurrence,
platformId: r.platform_id,
channelType: r.channel_type,
threadId: r.thread_id,
}),
});
log(`schedule_task: ${id} at ${processAfter}${recurrence ? ` (recurring: ${recurrence})` : ''}`);
return ok(`Task scheduled (id: ${id}, runs at: ${processAfter}${recurrence ? `, recurrence: ${recurrence}` : ''})`);
},
};
export const listTasks: McpToolDefinition = {
tool: {
name: 'list_tasks',
description:
'List scheduled tasks. Returns one row per series — the live (pending or paused) occurrence. The id shown is the series id, which is what update_task / cancel_task / pause_task / resume_task expect.',
inputSchema: {
type: 'object' as const,
properties: {
status: { type: 'string', description: 'Filter by status: pending or paused (default: both)' },
},
},
},
async handler(args) {
const status = args.status as string | undefined;
const db = getInboundDb();
// One row per series — the live (pending or paused) occurrence. Recurring
// tasks accumulate one completed row per firing plus one live follow-up;
// exposing the whole pile to the agent is noisy and confuses task identity
// ("which id do I cancel?"). The series_id is the stable handle.
//
// SQLite quirk: when MAX(seq) appears in the SELECT list of a GROUP BY
// query, the bare columns take values from the row that contains that max
// — that's how we pick "the latest live row per series" in one pass.
let rows;
if (status) {
rows = db
.prepare(
`SELECT series_id AS id, status, process_after, recurrence, content, MAX(seq) AS _seq
FROM messages_in
WHERE kind = 'task' AND status = ?
GROUP BY series_id
ORDER BY process_after ASC`,
)
.all(status);
} else {
rows = db
.prepare(
`SELECT series_id AS id, status, process_after, recurrence, content, MAX(seq) AS _seq
FROM messages_in
WHERE kind = 'task' AND status IN ('pending', 'paused')
GROUP BY series_id
ORDER BY process_after ASC`,
)
.all();
}
if ((rows as unknown[]).length === 0) return ok('No tasks found.');
const lines = (rows as Array<{ id: string; status: string; process_after: string | null; recurrence: string | null; content: string }>).map((r) => {
const content = JSON.parse(r.content);
const prompt = (content.prompt as string || '').slice(0, 80);
return `- ${r.id} [${r.status}] at=${r.process_after || 'now'} ${r.recurrence ? `recur=${r.recurrence} ` : ''}${prompt}`;
});
return ok(lines.join('\n'));
},
};
export const cancelTask: McpToolDefinition = {
tool: {
name: 'cancel_task',
description: 'Cancel a scheduled task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to cancel' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
// Write as a system action — host will update inbound.db
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'cancel_task', taskId }),
});
log(`cancel_task: ${taskId}`);
return ok(`Task cancellation requested: ${taskId}`);
},
};
export const pauseTask: McpToolDefinition = {
tool: {
name: 'pause_task',
description: 'Pause a scheduled task. It will not run until resumed.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to pause' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'pause_task', taskId }),
});
log(`pause_task: ${taskId}`);
return ok(`Task pause requested: ${taskId}`);
},
};
export const resumeTask: McpToolDefinition = {
tool: {
name: 'resume_task',
description: 'Resume a paused task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Task ID to resume' },
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'resume_task', taskId }),
});
log(`resume_task: ${taskId}`);
return ok(`Task resume requested: ${taskId}`);
},
};
export const updateTask: McpToolDefinition = {
tool: {
name: 'update_task',
description:
'Update a scheduled task. Pass the series id from list_tasks. Any field omitted is left unchanged. Use this instead of cancel + reschedule when adjusting an existing task.',
inputSchema: {
type: 'object' as const,
properties: {
taskId: { type: 'string', description: 'Series id of the task to update (as shown by list_tasks)' },
prompt: { type: 'string', description: 'New task prompt (optional)' },
recurrence: {
type: 'string',
description: 'New cron expression (optional). Pass empty string to clear and make the task one-shot.',
},
processAfter: {
type: 'string',
description:
`New ISO 8601 timestamp for the next run (optional). Accepts either UTC (ending in "Z" / "+00:00") or a naive local timestamp interpreted in the user's timezone.`,
},
script: {
type: 'string',
description: 'New pre-agent script (optional). Pass empty string to clear.',
},
},
required: ['taskId'],
},
},
async handler(args) {
const taskId = args.taskId as string;
if (!taskId) return err('taskId is required');
const update: Record<string, unknown> = { taskId };
if (typeof args.prompt === 'string') update.prompt = args.prompt;
if (typeof args.processAfter === 'string') {
try {
const d = parseZonedToUtc(args.processAfter, TIMEZONE);
if (Number.isNaN(d.getTime())) return err(`invalid processAfter: ${args.processAfter}`);
update.processAfter = d.toISOString();
} catch {
return err(`invalid processAfter: ${args.processAfter}`);
}
}
// Empty string clears recurrence/script; undefined leaves them as-is.
if (typeof args.recurrence === 'string') update.recurrence = args.recurrence === '' ? null : args.recurrence;
if (typeof args.script === 'string') update.script = args.script === '' ? null : args.script;
if (Object.keys(update).length === 1) return err('at least one field to update is required');
writeMessageOut({
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'system',
content: JSON.stringify({ action: 'update_task', ...update }),
});
log(`update_task: ${taskId}`);
return ok(`Task update requested: ${taskId}`);
},
};
registerTools([scheduleTask, listTasks, updateTask, cancelTask, pauseTask, resumeTask]);
+19 -10
View File
@@ -1,6 +1,6 @@
import { findByName, getAllDestinations, type DestinationEntry } from './destinations.js';
import { getPendingMessages, markProcessing, markCompleted, type MessageInRow } from './db/messages-in.js';
import { writeMessageOut } from './db/messages-out.js';
import { getPendingMessages, markProcessing, markCompleted, markScriptSkipped, type MessageInRow } from './db/messages-in.js';
import { hasIdenticalSend, writeMessageOut } from './db/messages-out.js';
import { getInboundDb, touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js';
import {
clearContinuation,
@@ -207,15 +207,15 @@ export async function runPollLoop(config: PollLoopConfig): Promise<void> {
// Without the scheduling module, the marker block is empty, `keep`
// falls back to `normalMessages`, and no gating happens.
let keep: MessageInRow[] = normalMessages;
let skipped: string[] = [];
let skipped: Array<{ id: string; reason: string }> = [];
// MODULE-HOOK:scheduling-pre-task:start
const { applyPreTaskScripts } = await import('./scheduling/task-script.js');
const preTask = await applyPreTaskScripts(normalMessages);
keep = preTask.keep;
skipped = preTask.skipped;
if (skipped.length > 0) {
markCompleted(skipped);
log(`Pre-task script skipped ${skipped.length} task(s): ${skipped.join(', ')}`);
markScriptSkipped(skipped);
log(`Pre-task script skipped ${skipped.length} task(s): ${skipped.map((s) => s.id).join(', ')}`);
}
// MODULE-HOOK:scheduling-pre-task:end
@@ -238,7 +238,7 @@ export async function runPollLoop(config: PollLoopConfig): Promise<void> {
});
// Process the query while concurrently polling for new messages
const skippedSet = new Set(skipped);
const skippedSet = new Set(skipped.map((s) => s.id));
const processingIds = ids.filter((id) => !commandIds.includes(id) && !skippedSet.has(id));
// Publish the batch's in_reply_to so MCP tools (send_message, send_file)
// can stamp it on outbound rows — needed for a2a return-path routing.
@@ -401,15 +401,15 @@ export async function processQuery(
// its script gate and always wakes the agent, defeating the gate.
// Mirrors the initial-batch hook above.
let keep = newMessages;
let skipped: string[] = [];
let skipped: Array<{ id: string; reason: string }> = [];
// MODULE-HOOK:scheduling-pre-task-followup:start
const { applyPreTaskScripts } = await import('./scheduling/task-script.js');
const preTask = await applyPreTaskScripts(newMessages);
keep = preTask.keep;
skipped = preTask.skipped;
if (skipped.length > 0) {
markCompleted(skipped);
log(`Pre-task script skipped ${skipped.length} follow-up task(s): ${skipped.join(', ')}`);
markScriptSkipped(skipped);
log(`Pre-task script skipped ${skipped.length} follow-up task(s): ${skipped.map((s) => s.id).join(', ')}`);
}
// MODULE-HOOK:scheduling-pre-task-followup:end
@@ -650,6 +650,15 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb
function sendToDestination(dest: DestinationEntry, body: string, routing: RoutingContext): void {
const platformId = dest.type === 'channel' ? dest.platformId! : dest.agentGroupId!;
const channelType = dest.type === 'channel' ? dest.channelType! : 'agent';
// Task fires: an explicitly-addressed final-text block is either the echo of
// an MCP send the agent already made this turn (drop it HERE, where the
// duplication originates) or the agent's only deliberate send (write it
// in_reply_to-null like the MCP path, or the host's task-fire suppression
// would discard it — zero delivery).
if (routing.taskFire && hasIdenticalSend(platformId, channelType, body)) {
log(`Dropping turn-final echo of an already-sent task message to ${dest.name}`);
return;
}
// Resolve thread_id per-destination from the most recent inbound message
// that came from this same channel+platform. In agent-shared sessions,
// different destinations have different thread contexts — using a single
@@ -657,7 +666,7 @@ function sendToDestination(dest: DestinationEntry, body: string, routing: Routin
const destRouting = resolveDestinationThread(channelType, platformId);
writeMessageOut({
id: generateId(),
in_reply_to: destRouting?.inReplyTo ?? routing.inReplyTo,
in_reply_to: destRouting?.inReplyTo ?? (routing.taskFire ? null : routing.inReplyTo),
kind: 'chat',
platform_id: platformId,
channel_type: channelType,
@@ -17,7 +17,7 @@ function log(msg: string): void {
// Code's interactive UI and would hang here).
//
// - CronCreate / CronDelete / CronList / ScheduleWakeup: we have durable
// scheduling via mcp__nanoclaw__schedule_task.
// scheduling via `ncl tasks`.
// - AskUserQuestion: SDK returns a placeholder instead of blocking on a
// real answer — we have mcp__nanoclaw__ask_user_question that persists
// the question and blocks on the real reply.
@@ -0,0 +1,72 @@
/**
* Container leg of the script-failure backoff chain, tested at unit level so
* the e2e suite doesn't need a live multi-sweep scenario for it:
*
* script error applyPreTaskScripts skips with reason 'error'
* markScriptSkipped acks `script-skip:error` in outbound.db
* (gated plain 'completed': the monitor working as designed).
*
* The host leg (ack FAILED run streak backoff) is pinned in
* src/db/session-db.test.ts and src/modules/scheduling/recurrence.test.ts
* both sides pin the literal 'script-skip:error'; if either renames it, its
* own test goes red.
*/
import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from '../db/connection.js';
import { getPendingMessages, markScriptSkipped } from '../db/messages-in.js';
import { applyPreTaskScripts } from './task-script.js';
beforeEach(() => {
initTestSessionDb();
});
afterEach(() => {
closeSessionDb();
});
function insertTask(id: string, script: string) {
getInboundDb()
.prepare(
`INSERT INTO messages_in (id, kind, timestamp, status, trigger, content)
VALUES (?, 'task', datetime('now'), 'pending', 1, ?)`,
)
.run(id, JSON.stringify({ prompt: 'monitor', script }));
}
const ackStatus = (id: string): string | undefined =>
(getOutboundDb().prepare('SELECT status FROM processing_ack WHERE message_id = ?').get(id) as { status: string } | undefined)
?.status;
describe('script-skip ack chain (container leg)', () => {
it('an erroring script skips with reason "error" and acks script-skip:error', async () => {
insertTask('t-err', 'echo boom >&2; exit 1');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(keep).toHaveLength(0);
expect(skipped).toEqual([{ id: 't-err', reason: 'error' }]);
markScriptSkipped(skipped);
expect(ackStatus('t-err')).toBe('script-skip:error');
});
it('a deliberate wakeAgent=false gate acks plain completed — never backs off', async () => {
insertTask('t-gated', 'echo \'{"wakeAgent": false}\'');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(keep).toHaveLength(0);
expect(skipped).toEqual([{ id: 't-gated', reason: 'gated' }]);
markScriptSkipped(skipped);
expect(ackStatus('t-gated')).toBe('completed');
});
it('wakeAgent=true keeps the task and enriches the prompt with script data', async () => {
insertTask('t-wake', 'echo \'{"wakeAgent": true, "data": {"alerts": 2}}\'');
const { keep, skipped } = await applyPreTaskScripts(getPendingMessages());
expect(skipped).toHaveLength(0);
expect(keep).toHaveLength(1);
expect(JSON.parse(keep[0].content).scriptOutput).toEqual({ alerts: 2 });
});
});
@@ -64,21 +64,26 @@ export async function runScript(script: string, taskId: string): Promise<ScriptR
});
}
/** Why a script gated its task: deliberate wakeAgent=false vs a broken script. */
export type ScriptSkipReason = 'gated' | 'error';
export interface TaskScriptOutcome {
keep: MessageInRow[];
skipped: string[];
skipped: Array<{ id: string; reason: ScriptSkipReason }>;
}
/**
* Run pre-task scripts for any task messages that carry one, serially.
* - Errors / missing output / wakeAgent=false task id added to `skipped`.
* - Errors / missing output / wakeAgent=false task id added to `skipped`,
* with the reason. The caller acks these as script-skips (not plain
* completions) so the host can count consecutive failures and back off.
* - wakeAgent=true content JSON is mutated to carry `scriptOutput`, so the
* formatter renders it into the prompt.
* Non-task messages and tasks without scripts pass through unchanged.
*/
export async function applyPreTaskScripts(messages: MessageInRow[]): Promise<TaskScriptOutcome> {
const keep: MessageInRow[] = [];
const skipped: string[] = [];
const skipped: Array<{ id: string; reason: ScriptSkipReason }> = [];
for (const msg of messages) {
if (msg.kind !== 'task') {
@@ -106,9 +111,9 @@ export async function applyPreTaskScripts(messages: MessageInRow[]): Promise<Tas
touchHeartbeat();
if (!result || !result.wakeAgent) {
const reason = result ? 'wakeAgent=false' : 'script error/no output';
log(`task ${msg.id} skipped: ${reason}`);
skipped.push(msg.id);
const reason: ScriptSkipReason = result ? 'gated' : 'error';
log(`task ${msg.id} skipped: ${reason === 'gated' ? 'wakeAgent=false' : 'script error/no output'}`);
skipped.push({ id: msg.id, reason });
continue;
}
+3 -3
View File
@@ -134,11 +134,11 @@ A personal AI assistant accessible via messaging, with minimal custom code.
### Scheduler
- Built-in scheduler runs on the host, spawns containers for task execution
- Custom `nanoclaw` MCP server (inside container) provides scheduling tools
- Tools: `schedule_task`, `list_tasks`, `pause_task`, `resume_task`, `cancel_task`, `send_message`
- `ncl tasks` provides scheduling commands
- Commands: `list`, `get`, `create`, `update`, `cancel`, `pause`, `resume`, `delete`, `run`, `append-log`
- Tasks stored in SQLite with run history
- Scheduler loop checks for due tasks every minute
- Tasks execute Claude Agent SDK in containerized group context
- Tasks execute in the agent group's system session
### Web Access
- Built-in WebSearch and WebFetch tools
+3 -20
View File
@@ -579,12 +579,7 @@ NanoClaw has a built-in scheduler that runs tasks as full agents in their group'
```
User: @Andy remind me every Monday at 9am to review the weekly metrics
Claude: [calls mcp__nanoclaw__schedule_task]
{
"prompt": "Send a reminder to review weekly metrics. Be encouraging!",
"schedule_type": "cron",
"schedule_value": "0 9 * * 1"
}
Claude: [runs ncl tasks create --prompt "Send a reminder to review weekly metrics. Be encouraging!" --process-after "2024-02-05T09:00:00" --recurrence "0 9 * * 1"]
Claude: Done! I'll remind you every Monday at 9am.
```
@@ -594,12 +589,7 @@ Claude: Done! I'll remind you every Monday at 9am.
```
User: @Andy at 5pm today, send me a summary of today's emails
Claude: [calls mcp__nanoclaw__schedule_task]
{
"prompt": "Search for today's emails, summarize the important ones, and send the summary to the group.",
"schedule_type": "once",
"schedule_value": "2024-01-31T17:00:00Z"
}
Claude: [runs ncl tasks create --prompt "Search for today's emails, summarize the important ones, and send the summary to the group." --process-after "2024-01-31T17:00:00"]
```
### Managing Tasks
@@ -620,18 +610,11 @@ From main channel:
### NanoClaw MCP (built-in)
The `nanoclaw` MCP server is created dynamically per agent call with the current group's context.
The `nanoclaw` MCP server is created dynamically per agent call with the current group's context. Scheduled task management lives in `ncl tasks`, not MCP.
**Available Tools:**
| Tool | Purpose |
|------|---------|
| `schedule_task` | Schedule a recurring or one-time task |
| `list_tasks` | Show tasks (group's tasks, or all if main) |
| `get_task` | Get task details and run history |
| `update_task` | Modify task prompt or schedule |
| `pause_task` | Pause a task |
| `resume_task` | Resume a paused task |
| `cancel_task` | Delete a task |
| `send_message` | Send a message to the group via its channel |
---
+8 -42
View File
@@ -634,52 +634,18 @@ destination is of type `agent`. `resolveRouting` maps it to a `messages_out` row
`channel_type: 'agent'` and `platform_id` set to the target agent group id; the host
validates the send and routes it into the target session's `inbound.db`.
#### schedule_task
#### ncl tasks
Schedule a one-shot or recurring task.
Schedule, inspect, and modify one-shot or recurring tasks.
```typescript
{
name: 'schedule_task',
params: {
prompt: string, // task prompt
processAfter: string, // ISO timestamp for first run
recurrence?: string, // cron expression (optional)
script?: string, // pre-agent script (optional)
}
}
```bash
ncl tasks create --prompt "..." --process-after "2026-01-15T09:00:00" --recurrence "0 9 * * *"
ncl tasks list
ncl tasks update <series_id> --prompt "..."
ncl tasks cancel <series_id>
```
Implementation: the container can't write host-owned `inbound.db`, so this writes a `messages_out` row with `kind: 'system'` and `action: 'schedule_task'` (`container/agent-runner/src/mcp-tools/scheduling.ts`). During delivery the host's action handler (`src/modules/scheduling/actions.ts``insertTask()` in `src/modules/scheduling/db.ts`) inserts the `kind: 'task'` row into `inbound.db` with `process_after` and optionally `recurrence`. The host sweep picks it up when due.
#### list_tasks
List active scheduled/recurring tasks.
```typescript
{
name: 'list_tasks',
params: {}
}
```
Implementation: a read, not a write — the container may read the read-only `inbound.db` mount directly. Returns one row per series (the live pending/paused occurrence): `SELECT series_id AS id, ... FROM messages_in WHERE kind = 'task' AND status IN ('pending','paused') GROUP BY series_id`. See `container/agent-runner/src/mcp-tools/scheduling.ts`.
#### cancel_task / pause_task / resume_task / update_task
Modify a scheduled task.
```typescript
{
name: 'cancel_task',
params: { taskId: string }
}
// pause_task: set status = 'paused' (new status value for recurring tasks)
// resume_task: set status = 'pending'
// update_task: merge { prompt?, recurrence?, processAfter?, script? } into the live row
```
Implementation: all four are sent as system actions (`messages_out`, `kind: 'system'`, `action: 'cancel_task' | 'pause_task' | 'resume_task' | 'update_task'`) — the container never writes `inbound.db`. The host's handlers in `src/modules/scheduling/actions.ts` apply the change against `inbound.db` via `src/modules/scheduling/db.ts`: cancel/pause/resume flip status on the live row(s); update_task reads current content, merges supplied fields, and writes back. All four match by `(id = ? OR series_id = ?) AND kind='task' AND status IN ('pending','paused')`, so they reach the live next occurrence of a recurring task even when the agent passes the original (now-completed) id.
Implementation: the host writes `messages_in` task rows into the agent group's system session (`thread_id = system:tasks`). The host sweep wakes that system-session container when a task is due. The task agent chooses its destination at fire time by emitting `<message to="name">...</message>` or using `send_message`.
#### create_agent
+1 -1
View File
@@ -32,7 +32,7 @@ flowchart TB
direction TB
PollLoop["Poll Loop<br/>(container/agent-runner)"]
Provider["Agent providers<br/>(claude — the only one registered in trunk;<br/>opencode ships via the /add-opencode skill)"]
MCP["MCP Tools<br/>send_message, send_file, edit_message,<br/>add_reaction, send_card, ask_user_question,<br/>schedule_task, create_agent,<br/>install_packages, add_mcp_server"]
MCP["MCP Tools<br/>send_message, send_file, edit_message,<br/>add_reaction, send_card, ask_user_question,<br/>create_agent, install_packages, add_mcp_server<br/>CLI: ncl tasks"]
Skills["Container Skills<br/>(container/skills/)"]
InDB[("inbound.db<br/>host writes<br/>even seq<br/>messages_in<br/>delivered<br/>destinations<br/>session_routing")]
OutDB[("outbound.db<br/>container writes<br/>odd seq<br/>messages_out<br/>processing_ack<br/>session_state<br/>container_state<br/>heartbeat file")]
+4 -7
View File
@@ -904,13 +904,10 @@ MCP tools write to the container's own `outbound.db`. Anything that needs a chan
(There is no `send_to_agent` tool — agent-to-agent is `send_message` to an `agent` destination.)
**Scheduling** (all emit `kind: 'system'` actions except the read-only `list_tasks`):
| Tool | What it does |
|------|-------------|
| `schedule_task` | `action: 'schedule_task'`; host inserts the `kind: 'task'` `messages_in` row with `process_after` + optional `recurrence` |
| `list_tasks` | Read `inbound.db` (read-only) — one row per series: `kind = 'task' AND status IN ('pending','paused') GROUP BY series_id` |
| `pause_task` / `resume_task` / `cancel_task` / `update_task` | matching `action`; host updates the live `messages_in` row(s) |
**Scheduling**: scheduled-task management is not an MCP surface — it lives on
`ncl tasks` (create/list/get/update/cancel/pause/resume/run/append-log). Due
task rows live in the agent group's system session and are woken by the host
sweep.
**Central-DB / self-modification** (`kind: 'system'` actions; host authorizes, often via admin approval):
+52
View File
@@ -0,0 +1,52 @@
# `ncl tasks` migration
## Detect
If an agent mentions `schedule_task`, `list_tasks`, `update_task`, `cancel_task`, `pause_task`, or `resume_task`, it is using the old scheduling MCP surface.
A subtler symptom of a stale container image: the agent reports a task as scheduled, but `ncl tasks list` shows nothing and the host log has `Unknown system action` — the old image's `schedule_task` call is acknowledged in-container and then dropped by the new host. The fix below (rebuild + restart) resolves it.
## Why
Scheduling moved to `ncl tasks`. New tasks are stored in a per-agent-group system session and run there, so a scheduled task does not wake an existing chat session. When it fires, the agent must choose the delivery destination explicitly.
## Fix
Rebuild and restart agent containers so they load the updated MCP tool list and instructions:
```bash
./container/build.sh
launchctl kickstart -k gui/$(id -u)/com.nanoclaw
```
On Linux, restart with `systemctl --user restart nanoclaw`.
Use:
```bash
ncl tasks list
ncl tasks create --group <agent_group_id> --prompt "..." --process-after "2026-01-15T09:00:00" --recurrence "0 9 * * *"
ncl tasks update --id <series_id> --prompt "..."
ncl tasks cancel --id <series_id>
```
## Verify
Run `ncl tasks list`. New task rows should show a system `session_id`, not the chat session that requested the task.
## Legacy tasks (scheduled before this update)
Tasks created through the old MCP tools live in the **chat session** that created them, not in a per-series system session. They are unaffected by this update: they keep firing and delivering exactly as before. Two things to know:
- An agent's own `ncl tasks list` (group scope) shows only its group's task rows; from the **host**, unscoped `ncl tasks list` enumerates everything, and `--session <id>` narrows to one session — that is how you find and manage legacy rows (`ncl tasks cancel --session <chat_session_id> --all` to clear a chat session's tasks).
- The `messages_in` status enum now includes `cancelled` (cancel marks the row and clears its recurrence rather than deleting it). Custom code that exhaustively switches on task status needs the new arm.
## Rollback
Order matters:
1. Remove tasks created through `ncl tasks` (`ncl tasks list` / `delete`) — they live in per-series system sessions the old code doesn't know about.
2. **Wait one sweep (≤60s)** so the host closes the now-empty task sessions.
3. Then revert the update and rebuild the container image.
Reverting before the task sessions are collected leaves system sessions behind that the old `findSessionByAgentGroup` (which has no system-session exclusion) can resolve as the group's session — mis-routing agent-to-agent messages into a dead task thread.
+26 -1
View File
@@ -15,7 +15,7 @@ vi.mock('./log.js', () => ({
}));
import { composeGroupClaudeMd } from './claude-md-compose.js';
import { ensureContainerConfig } from './db/container-configs.js';
import { ensureContainerConfig, updateContainerConfigScalars } from './db/container-configs.js';
import { closeDb, createAgentGroup, initTestDb, runMigrations } from './db/index.js';
import { PERSONA_PREPEND_FILE } from './group-persona.js';
import type { AgentGroup } from './types.js';
@@ -91,3 +91,28 @@ describe('composeGroupClaudeMd persona prepend', () => {
expect(fs.existsSync(path.join(GROUPS_DIR, ag.folder, '.claude-fragments', 'persona.md'))).toBe(false);
});
});
describe('composeGroupClaudeMd scheduling instructions (ncl tasks reach-in)', () => {
// Red-on-delete guard for the `scheduling`/`cli` exclusion at the
// module-fragment loop: the agent is taught `ncl tasks` iff it has ncl.
it('imports module-scheduling.md at the default cli_scope', () => {
const ag = group('ag-sched', 'sched-group');
seed(ag);
composeGroupClaudeMd(ag);
expect(importsOf(ag.folder)).toContain('@./.claude-fragments/module-scheduling.md');
});
it('excludes module-scheduling.md (and module-cli.md) when cli_scope is disabled', () => {
const ag = group('ag-sched-off', 'sched-group-off');
seed(ag);
updateContainerConfigScalars(ag.id, { cli_scope: 'disabled' });
composeGroupClaudeMd(ag);
const imports = importsOf(ag.folder);
expect(imports).not.toContain('@./.claude-fragments/module-scheduling.md');
expect(imports).not.toContain('@./.claude-fragments/module-cli.md');
});
});
+6 -4
View File
@@ -80,10 +80,12 @@ export function composeGroupClaudeMd(group: AgentGroup): void {
}
}
// Built-in module fragments — every MCP tool source file that ships a
// Built-in module fragments — every MCP/CLI module that ships a
// sibling `<name>.instructions.md`. These describe how the agent should
// use that module's MCP tools (schedule_task, install_packages, etc.).
// Skip cli.instructions.md when cli_scope is disabled.
// use that module's tools (`ncl tasks`, install_packages, etc.).
// Skip ncl-dependent instructions when cli_scope is disabled. `scheduling`
// teaches `ncl tasks`, so it is just as dead as `cli` itself when the agent
// has no ncl — dispatch rejects every cli_request and ncl is excluded.
const cliDisabled = configRow?.cli_scope === 'disabled';
const mcpToolsHostDir = path.join(process.cwd(), MCP_TOOLS_HOST_SUBPATH);
if (fs.existsSync(mcpToolsHostDir)) {
@@ -91,7 +93,7 @@ export function composeGroupClaudeMd(group: AgentGroup): void {
const match = entry.match(/^(.+)\.instructions\.md$/);
if (!match) continue;
const moduleName = match[1];
if (moduleName === 'cli' && cliDisabled) continue;
if ((moduleName === 'cli' || moduleName === 'scheduling') && cliDisabled) continue;
desired.set(`module-${moduleName}.md`, {
type: 'symlink',
content: `${SHARED_MCP_TOOLS_CONTAINER_BASE}/${entry}`,
+23 -68
View File
@@ -115,6 +115,15 @@ register({
handler: async (args) => ({ echo: args }),
});
register({
name: 'tasks-list',
description: 'test command (tasks resource)',
resource: 'tasks',
access: 'open',
parseArgs: (raw) => raw,
handler: async (args) => ({ echo: args }),
});
register({
name: 'wirings-list',
description: 'test command (wirings resource — not allowed)',
@@ -136,33 +145,6 @@ register({
},
});
register({
name: 'roles-grant',
description: 'approval command on a non-scoped resource (global blast radius)',
resource: 'roles',
access: 'approval',
parseArgs: (raw) => raw,
handler: async () => ({}),
});
register({
name: 'members-add-gated',
description: 'approval command on a scoped resource',
resource: 'members',
access: 'approval',
parseArgs: (raw) => raw,
handler: async () => ({}),
});
register({
name: 'groups-update',
description: 'approval command on the groups resource (id = agent group)',
resource: 'groups',
access: 'approval',
parseArgs: (raw) => raw,
handler: async () => ({}),
});
// Commands that return data shaped like real resources (for post-handler filtering tests)
register({
name: 'groups-list-data',
@@ -245,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,
@@ -391,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' });
@@ -511,47 +507,6 @@ describe('CLI scope enforcement', () => {
expect(approvalState.requestApproval).toHaveBeenCalledTimes(1);
});
// --- Approver blast radius (D1) ---
it('holds on non-scoped resources carry approverScope global (roles grant)', async () => {
mockGetContainerConfig.mockReturnValue({ cli_scope: 'global' });
mockGetSession.mockReturnValue({ id: 's1', agent_group_id: 'g1', messaging_group_id: 'mg1' });
mockGetAgentGroup.mockReturnValue({ id: 'g1', name: 'Group One' });
const resp = await dispatch(
{ id: '1', command: 'roles-grant', args: { user: 'telegram:mallory', role: 'owner' } },
agentCtx(),
);
expect(resp.ok).toBe(false);
expect(approvalState.requestApproval).toHaveBeenCalledTimes(1);
expect(approvalState.requestApproval.mock.calls[0][0]).toMatchObject({ approverScope: 'global' });
});
it('holds on scoped resources pinned to the caller stay approverScope group', async () => {
mockGetContainerConfig.mockReturnValue({ cli_scope: 'group' });
mockGetSession.mockReturnValue({ id: 's1', agent_group_id: 'g1', messaging_group_id: 'mg1' });
mockGetAgentGroup.mockReturnValue({ id: 'g1', name: 'Group One' });
const resp = await dispatch({ id: '1', command: 'members-add-gated', args: { user: 'telegram:new' } }, agentCtx());
expect(resp.ok).toBe(false);
expect(approvalState.requestApproval).toHaveBeenCalledTimes(1);
expect(approvalState.requestApproval.mock.calls[0][0]).toMatchObject({ approverScope: 'group' });
});
it('holds on scoped resources targeting another group escalate to approverScope global', async () => {
mockGetContainerConfig.mockReturnValue({ cli_scope: 'global' });
mockGetSession.mockReturnValue({ id: 's1', agent_group_id: 'g1', messaging_group_id: 'mg1' });
mockGetAgentGroup.mockReturnValue({ id: 'g1', name: 'Group One' });
const resp = await dispatch({ id: '1', command: 'groups-update', args: { id: 'g2' } }, agentCtx());
expect(resp.ok).toBe(false);
expect(approvalState.requestApproval).toHaveBeenCalledTimes(1);
expect(approvalState.requestApproval.mock.calls[0][0]).toMatchObject({ approverScope: 'global' });
});
// --- Post-handler filtering ---
it('group: groups list filters out other groups', async () => {
+1 -21
View File
@@ -13,32 +13,13 @@ import { registerApprovalHandler, requestApproval } from '../modules/approvals/i
import type { CallerContext, ErrorCode, RequestFrame, ResponseFrame } from './frame.js';
import { getResource } from './crud.js';
import { listVerbs, renderVerbHelp } from './help-render.js';
import { GROUP_SCOPE_RESOURCES, listCommands, lookup, type CommandDef } from './registry.js';
import { GROUP_SCOPE_RESOURCES, listCommands, lookup } from './registry.js';
type DispatchOptions = {
/** True when a command is being replayed after approval. */
approved?: boolean;
};
/**
* Blast radius of a held command, for the hold's approver rule (D1): a
* mutation of a non-group-scoped resource (roles, users, wirings,
* messaging-groups, policies) or one explicitly targeting another agent
* group needs an owner or global admin to approve; a scoped admin's click
* is rejected. GROUP_SCOPE_RESOURCES anchors rows to one agent group, so its
* held mutations default to group-local blast radius.
*/
function approverScopeFor(
cmd: CommandDef,
args: Record<string, unknown>,
callerAgentGroupId: string,
): 'group' | 'global' {
if (!cmd.resource || !GROUP_SCOPE_RESOURCES.has(cmd.resource)) return 'global';
const groupRefs = [args.agent_group_id, args.group];
if (cmd.resource === 'groups' || cmd.resource === 'destinations') groupRefs.push(args.id);
return groupRefs.some((v) => v !== undefined && v !== callerAgentGroupId) ? 'global' : 'group';
}
export async function dispatch(
req: RequestFrame,
ctx: CallerContext,
@@ -165,7 +146,6 @@ export async function dispatch(
payload: { frame: { id: req.id, command: req.command, args: req.args }, callerContext: ctx },
title: `CLI: ${req.command}`,
question: `Agent "${agentName}" wants to run:\n\`ncl ${req.command}${argSummary ? ' ' + argSummary : ''}\``,
approverScope: approverScopeFor(cmd, req.args, ctx.agentGroupId),
});
return err(req.id, 'approval-pending', 'Approval request sent to admin. You will be notified of the result.');
+84
View File
@@ -0,0 +1,84 @@
/**
* Compact aligned-table renderer for `ncl tasks list` (human mode only).
*
* A task series is a recurring job; each fire is a run. This surfaces the run history
* the raw row hides run count, last/next fire, schedule as an aligned table.
* Driven by the enriched rows from listTasks (tasks.ts); the --json path is
* untouched. `now` is injectable so the relative times are testable.
*/
interface TaskListRow {
series_id: string;
schedule?: string | null;
runs?: number;
failed_runs?: number;
last_run?: string | null;
next_run?: string | null;
status?: string;
log?: string | null;
created_at?: string | null;
prompt?: string | null;
}
const COLS = ['SERIES', 'SCHEDULE', 'RUNS', 'FAILED', 'LAST RUN', 'NEXT RUN', 'STATUS', 'AGE', 'PROMPT'] as const;
function parseMs(iso: string): number {
return Date.parse(/[Z+]|[+-]\d\d:\d\d$/.test(iso) ? iso : iso + 'Z');
}
/** "1m ago" / "in 30s" — coarse, human relative time. */
function duration(ms: number): string {
const s = Math.abs(ms) / 1000;
if (s < 60) return `${Math.round(s)}s`;
if (s < 3600) return `${Math.round(s / 60)}m`;
if (s < 86400) return `${Math.round(s / 3600)}h`;
return `${Math.round(s / 86400)}d`;
}
function lastRun(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
if (Number.isNaN(t)) return iso;
return `${duration(now - t)} ago`;
}
function nextRun(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
if (Number.isNaN(t)) return iso;
return t <= now ? 'due' : `in ${duration(t - now)}`;
}
/** AGE — how long since the series was created. */
function age(iso: string | null | undefined, now: number): string {
if (!iso) return '-';
const t = parseMs(iso);
return Number.isNaN(t) ? '-' : duration(now - t);
}
function clip(s: string | null | undefined, n: number): string {
const v = (s ?? '').replace(/\s+/g, ' ').trim();
return v.length > n ? v.slice(0, n - 1) + '…' : v;
}
export function formatTasksTable(rows: TaskListRow[], now: number = Date.now()): string {
if (!rows.length) return 'No tasks.';
const body = rows.map((r) => [
r.series_id, // full id, copy-pasteable into `ncl tasks get --id <…>`
r.schedule || 'once',
String(r.runs ?? 0),
String(r.failed_runs ?? 0),
lastRun(r.last_run, now),
nextRun(r.next_run, now),
r.status ?? '-',
age(r.created_at, now),
clip(r.prompt, 40),
]);
const widths = COLS.map((c, i) => Math.max(c.length, ...body.map((row) => row[i].length)));
const line = (cells: string[]) =>
cells
.map((c, i) => c.padEnd(widths[i]))
.join(' ')
.trimEnd();
return [line([...COLS]), ...body.map(line)].join('\n');
}
+1 -1
View File
@@ -15,7 +15,7 @@ import type { CallerContext } from './frame.js';
* consumed by both dispatch enforcement and `ncl help` filtering, so the
* agent is never shown a resource the gate would reject (or vice versa).
*/
export const GROUP_SCOPE_RESOURCES = new Set(['groups', 'sessions', 'destinations', 'members']);
export const GROUP_SCOPE_RESOURCES = new Set(['groups', 'sessions', 'destinations', 'members', 'tasks']);
export type Access = 'open' | 'approval' | 'hidden';
-18
View File
@@ -48,24 +48,6 @@ registerResource({
},
{ name: 'title', type: 'string', description: 'Card title shown to the admin.' },
{ name: 'options_json', type: 'json', description: 'Card button options as JSON array.' },
{
name: 'approver_user_id',
type: 'string',
description: 'Named approver (exclusive) or the admin the card was delivered to (admins-of-scope).',
},
{
name: 'approver_rule',
type: 'string',
description: 'Who may resolve: only the named approver, or the admin chain of the anchoring group.',
enum: ['exclusive', 'admins-of-scope'],
},
{
name: 'approver_scope',
type: 'string',
description: "Blast radius: 'global' holds require an owner or global admin to resolve.",
enum: ['group', 'global'],
},
{ name: 'dedup_key', type: 'string', description: 'In-flight dedup key (e.g. sender admission per chat+sender).' },
],
operations: { list: 'open', get: 'open' },
});
+30 -1
View File
@@ -58,10 +58,39 @@ registerResource({
type: 'string',
description: "The target's ID — messaging_groups.id for channels, agent_groups.id for agents.",
},
{ name: 'channel_type', type: 'string', description: 'Resolved channel type for channel destinations.' },
{ name: 'display_name', type: 'string', description: 'Resolved chat title or agent name.' },
{ name: 'created_at', type: 'string', description: 'Auto-set.' },
],
operations: { list: 'open' },
operations: {},
customOperations: {
list: {
access: 'open',
description: 'List destinations with resolved channel/title labels.',
handler: async (args) => {
const agentGroupId = (args.agent_group_id as string | undefined) ?? (args.id as string | undefined);
const params: unknown[] = [];
const where = agentGroupId ? 'WHERE ad.agent_group_id = ?' : '';
if (agentGroupId) params.push(agentGroupId);
return getDb()
.prepare(
`SELECT
ad.agent_group_id,
ad.local_name,
ad.target_type,
ad.target_id,
CASE WHEN ad.target_type = 'channel' THEN mg.channel_type ELSE NULL END AS channel_type,
CASE WHEN ad.target_type = 'channel' THEN mg.name ELSE ag.name END AS display_name,
ad.created_at
FROM agent_destinations ad
LEFT JOIN messaging_groups mg ON ad.target_type = 'channel' AND ad.target_id = mg.id
LEFT JOIN agent_groups ag ON ad.target_type = 'agent' AND ad.target_id = ag.id
${where}
ORDER BY ad.agent_group_id, ad.local_name`,
)
.all(...params);
},
},
add: {
access: 'approval',
description: 'Add a destination for an agent. Use --agent-group-id, --local-name, --target-type, --target-id.',
+6 -7
View File
@@ -109,13 +109,10 @@ describe('groups CLI delete cascades dependent rows (#2525)', () => {
VALUES (?, ?, 'req-1', 'cli_command', '{}', ?, ?, 'pending', '', '[]')`,
).run('pa-1', SID, now(), GID);
// Sessionless sender-admission hold anchored to the group (the folded
// pending_sender_approvals shape) — covered by the agent_group_id leg of
// the pending_approvals cascade.
db.prepare(
`INSERT INTO pending_approvals (approval_id, session_id, request_id, action, payload, created_at, agent_group_id, status, title, options_json, dedup_key)
VALUES (?, NULL, 'req-2', 'sender_admit', '{}', ?, ?, 'pending', '', '[]', 'sender_admit:mg:tg:99')`,
).run('pa-2', now(), GID);
`INSERT INTO pending_sender_approvals (id, messaging_group_id, agent_group_id, sender_identity, sender_name, original_message, approver_user_id, created_at)
VALUES ('psa-1', ?, ?, 'tg:99', 'them', '{}', ?, ?)`,
).run(MGID, GID, UID, now());
db.prepare(
`INSERT INTO pending_channel_approvals (messaging_group_id, agent_group_id, original_message, approver_user_id, created_at)
@@ -151,9 +148,10 @@ describe('groups CLI delete cascades dependent rows (#2525)', () => {
expect(data.removed).toMatchObject({
sessions: 1,
pending_questions: 1,
pending_approvals: 2,
pending_approvals: 1,
agent_destinations_owned: 1,
agent_destinations_pointing: 0,
pending_sender_approvals: 1,
pending_channel_approvals: 1,
messaging_group_agents: 1,
agent_group_members: 1,
@@ -169,6 +167,7 @@ describe('groups CLI delete cascades dependent rows (#2525)', () => {
count('SELECT COUNT(*) AS c FROM pending_approvals WHERE agent_group_id = ? OR session_id = ?', GID, SID),
).toBe(0);
expect(count('SELECT COUNT(*) AS c FROM agent_destinations WHERE agent_group_id = ?', GID)).toBe(0);
expect(count('SELECT COUNT(*) AS c FROM pending_sender_approvals WHERE agent_group_id = ?', GID)).toBe(0);
expect(count('SELECT COUNT(*) AS c FROM pending_channel_approvals WHERE agent_group_id = ?', GID)).toBe(0);
expect(count('SELECT COUNT(*) AS c FROM messaging_group_agents WHERE agent_group_id = ?', GID)).toBe(0);
expect(count('SELECT COUNT(*) AS c FROM agent_group_members WHERE agent_group_id = ?', GID)).toBe(0);
+4
View File
@@ -124,6 +124,7 @@ registerResource({
pending_approvals: 0,
agent_destinations_owned: 0,
agent_destinations_pointing: 0,
pending_sender_approvals: 0,
pending_channel_approvals: 0,
messaging_group_agents: 0,
agent_group_members: 0,
@@ -152,6 +153,9 @@ registerResource({
.run(groupId, groupId).changes;
}
counts.sessions = db.prepare('DELETE FROM sessions WHERE agent_group_id = ?').run(groupId).changes;
counts.pending_sender_approvals = db
.prepare('DELETE FROM pending_sender_approvals WHERE agent_group_id = ?')
.run(groupId).changes;
counts.pending_channel_approvals = db
.prepare('DELETE FROM pending_channel_approvals WHERE agent_group_id = ?')
.run(groupId).changes;
+1
View File
@@ -14,3 +14,4 @@ import './user-dms.js';
import './dropped-messages.js';
import './approvals.js';
import './sessions.js';
import './tasks.js';
+575
View File
@@ -0,0 +1,575 @@
import Database from 'better-sqlite3';
import fs from 'fs';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('../../config.js', async () => {
const actual = await vi.importActual('../../config.js');
return {
...actual,
DATA_DIR: '/tmp/nanoclaw-test-cli-tasks',
GROUPS_DIR: '/tmp/nanoclaw-test-cli-tasks/groups',
TIMEZONE: 'UTC',
};
});
vi.mock('../../container-runner.js', () => ({
wakeContainer: vi.fn().mockResolvedValue(undefined),
isContainerRunning: vi.fn().mockReturnValue(false),
getActiveContainerCount: vi.fn().mockReturnValue(0),
killContainer: vi.fn(),
}));
const TEST_DIR = '/tmp/nanoclaw-test-cli-tasks';
import { initTestDb, closeDb, runMigrations, createAgentGroup } from '../../db/index.js';
import { createSession, findSessionByAgentGroup, getSessionsByAgentGroup, taskThreadId } from '../../db/sessions.js';
import { countDueMessages } from '../../db/session-db.js';
import { inboundDbPath, initSessionFolder } from '../../session-manager.js';
import { dispatch } from '../dispatch.js';
import { formatTasksTable } from '../format-tasks.js';
import type { CallerContext } from '../frame.js';
import './tasks.js';
import '../commands/index.js'; // registers tasks-help for the help-topic test
function now(): string {
return new Date().toISOString();
}
function createGroup(id: string): void {
createAgentGroup({ id, name: id, folder: id, agent_provider: null, created_at: now() });
}
function createChatSession(group: string, id: string): void {
createSession({
id,
agent_group_id: group,
messaging_group_id: null,
thread_id: null,
agent_provider: null,
status: 'active',
container_status: 'stopped',
last_active: null,
created_at: now(),
});
initSessionFolder(group, id);
}
function agentCtx(group = 'ag-1', session = 'chat-1'): CallerContext {
return { caller: 'agent', agentGroupId: group, sessionId: session, messagingGroupId: 'mg-1' };
}
describe('tasks CLI resource', () => {
beforeEach(() => {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
fs.mkdirSync(TEST_DIR, { recursive: true });
const db = initTestDb();
runMigrations(db);
createGroup('ag-1');
createGroup('ag-2');
createChatSession('ag-1', 'chat-1');
createChatSession('ag-2', 'chat-2');
});
afterEach(() => {
closeDb();
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
});
it('create writes the task into the group system session, not the caller chat session', async () => {
const resp = await dispatch(
{
id: 'req-1',
command: 'tasks-create',
args: { prompt: 'send a briefing', process_after: '2026-01-15T09:00:00Z' },
},
agentCtx(),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
const created = resp.data as { series_id: string; session_id: string };
expect(created.session_id).not.toBe('chat-1');
// The task lands in its own isolated per-series session, not the chat session.
const sessions = getSessionsByAgentGroup('ag-1');
const taskSession = sessions.find((s) => s.id === created.session_id);
expect(taskSession?.thread_id).toBe(taskThreadId(created.series_id));
const chatDb = new Database(inboundDbPath('ag-1', 'chat-1'), { readonly: true });
expect(chatDb.prepare("SELECT COUNT(*) AS count FROM messages_in WHERE kind = 'task'").get()).toEqual({
count: 0,
});
chatDb.close();
const systemDb = new Database(inboundDbPath('ag-1', created.session_id), { readonly: true });
const row = systemDb.prepare("SELECT content FROM messages_in WHERE kind = 'task'").get() as { content: string };
const content = JSON.parse(row.content);
expect(content).toMatchObject({ originSessionId: 'chat-1' });
expect(content.prompt).toContain('send a briefing');
expect(content.prompt).toContain(`tasks/${created.series_id}.md`); // log-path hint injected
systemDb.close();
});
it('tasks-list attaches a server-rendered human table (so the container agent gets it too)', async () => {
await dispatch(
{
id: 'c',
command: 'tasks-create',
args: { prompt: 'x', name: 'briefing', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx(),
);
const resp = await dispatch({ id: 'l', command: 'tasks-list', args: {} }, agentCtx());
expect(resp.ok).toBe(true);
if (!resp.ok) return;
// Red-on-delete guard for the dispatch wiring: the host renders format-tasks
// once and ships it as `human`, so the Bun container prints the aligned
// table instead of a raw column dump (it cannot import the host formatter).
expect(resp.human).toBeDefined();
expect(resp.human).toMatch(/SERIES\s+SCHEDULE\s+RUNS\s+FAILED\s+LAST RUN\s+NEXT RUN/);
expect(resp.human).toContain('briefing-');
});
it('recurrence more frequent than 4x/day is refused with the quota warning', async () => {
const resp = await dispatch(
{ id: 'c', command: 'tasks-create', args: { prompt: 'x', name: 'spam', recurrence: '*/2 * * * *' } },
agentCtx(),
);
expect(resp.ok).toBe(false);
if (!resp.ok) {
expect(resp.error.message).toContain('this task has not been scheduled');
expect(resp.error.message).toContain('ncl tasks create --help');
expect(resp.error.message).toContain('--dangerously-override-recurrence-limit');
}
});
it('exactly 4 fires/day passes; the override flag bypasses the limit', async () => {
const four = await dispatch(
{ id: 'c4', command: 'tasks-create', args: { prompt: 'x', name: 'four', recurrence: '0 0,6,12,18 * * *' } },
agentCtx(),
);
expect(four.ok).toBe(true);
const overridden = await dispatch(
{
id: 'co',
command: 'tasks-create',
args: { prompt: 'x', name: 'fast', recurrence: '*/30 * * * *', dangerously_override_recurrence_limit: true },
},
agentCtx(),
);
expect(overridden.ok).toBe(true);
});
it('a --script gate exempts frequent recurrence — the sanctioned monitor pattern', async () => {
const scripted = await dispatch(
{
id: 'cs',
command: 'tasks-create',
args: {
prompt: 'triage queue',
name: 'watch',
recurrence: '*/10 * * * *',
script: 'echo {"wakeAgent": false}',
},
},
agentCtx(),
);
expect(scripted.ok).toBe(true);
// update --recurrence on a task that already has a script: also exempt.
if (!scripted.ok) return;
const seriesId = (scripted.data as { series_id: string }).series_id;
const upd = await dispatch(
{ id: 'us', command: 'tasks-update', args: { id: seriesId, recurrence: '*/5 * * * *' } },
agentCtx(),
);
expect(upd.ok).toBe(true);
// …but clearing the script in the same update re-arms the guard.
const cleared = await dispatch(
{ id: 'uc', command: 'tasks-update', args: { id: seriesId, recurrence: '*/5 * * * *', script: 'none' } },
agentCtx(),
);
expect(cleared.ok).toBe(false);
});
it('the limit also guards update --recurrence (no create-slow-then-update bypass)', async () => {
const created = await dispatch(
{ id: 'c', command: 'tasks-create', args: { prompt: 'x', name: 'sneak', recurrence: '0 9 * * *' } },
agentCtx(),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const seriesId = (created.data as { series_id: string }).series_id;
const upd = await dispatch(
{ id: 'u', command: 'tasks-update', args: { id: seriesId, recurrence: '* * * * *' } },
agentCtx(),
);
expect(upd.ok).toBe(false);
if (!upd.ok) expect(upd.error.message).toContain('this task has not been scheduled');
});
it('tasks create --help carries the script contract and the frequency-limit caveat', async () => {
// --help and `tasks help create` render the same deep verb help.
const resp = await dispatch({ id: 'h', command: 'tasks-create', args: { help: true } }, agentCtx());
expect(resp.ok).toBe(true);
if (resp.ok) {
const text = resp.data as string;
expect(text).toContain('wakeAgent');
expect(text).toContain('Frequency limit');
}
});
it('agent-shared lookup skips the task system session', async () => {
await dispatch(
{
id: 'req-1',
command: 'tasks-create',
args: { prompt: 'send a briefing', process_after: '2026-01-15T09:00:00Z' },
},
agentCtx(),
);
expect(findSessionByAgentGroup('ag-1')?.id).toBe('chat-1');
});
it('group-scoped agents cannot list tasks from another group session', async () => {
const resp = await dispatch(
{ id: 'req-1', command: 'tasks-list', args: { session: 'chat-2' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) {
expect(resp.error.code).toBe('handler-error');
expect(resp.error.message).toContain('session not found');
}
});
it('--name yields a short, readable, fs/thread-safe id', async () => {
const r = await dispatch(
{
id: 'rn',
command: 'tasks-create',
args: { prompt: 'x', name: 'Morning Joke!!', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const id = (r.data as { series_id: string }).series_id;
expect(id).toMatch(/^morning-joke-[0-9a-f]{4}$/);
expect(id).toMatch(/^[a-z0-9-]+$/); // safe as thread suffix / filename / --id
});
it('no name yields a t-<hex> id', async () => {
const r = await dispatch(
{ id: 'rnn', command: 'tasks-create', args: { prompt: 'x', process_after: '2999-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
expect((r.data as { series_id: string }).series_id).toMatch(/^t-[0-9a-f]{6}$/);
});
it('recurring create derives the first run from the cron grid when --process-after is omitted', async () => {
const r = await dispatch(
{ id: 'rec', command: 'tasks-create', args: { prompt: 'x', name: 'nightly', recurrence: '0 9 * * 1-5' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const task = r.data as { process_after: string; recurrence: string };
expect(task.recurrence).toBe('0 9 * * 1-5');
// First fire snapped onto the cron grid (TIMEZONE=UTC in this suite).
const firstRun = new Date(task.process_after);
expect(Number.isNaN(firstRun.getTime())).toBe(false);
expect(firstRun.getUTCHours()).toBe(9);
expect(firstRun.getTime()).toBeGreaterThan(Date.now());
});
it('one-shot create still requires --process-after (nothing to derive it from)', async () => {
const r = await dispatch(
{ id: 'os', command: 'tasks-create', args: { prompt: 'x', name: 'once' } },
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(false);
if (!r.ok) expect(r.error.message).toContain('--process-after is required');
});
it('run queues an extra immediate occurrence without consuming the scheduled one', async () => {
const created = await dispatch(
{
id: 'c',
command: 'tasks-create',
args: { prompt: 'x', name: 'pingable', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { series_id, session_id } = created.data as { series_id: string; session_id: string };
const run = await dispatch({ id: 'r', command: 'tasks-run', args: { id: series_id } }, agentCtx('ag-1', 'chat-1'));
expect(run.ok).toBe(true);
if (!run.ok) return;
const fired = run.data as { series_id: string; row_id: string; status: string };
expect(fired.series_id).toBe(series_id);
expect(fired.row_id).not.toBe(series_id);
expect(fired.status).toBe('pending');
const db = new Database(inboundDbPath('ag-1', session_id), { readonly: true });
const pending = db
.prepare(
"SELECT id, recurrence, process_after FROM messages_in WHERE kind = 'task' AND status = 'pending' AND series_id = ?",
)
.all(series_id) as Array<{ id: string; recurrence: string | null; process_after: string }>;
db.close();
// Original scheduled row + the new run-now occurrence both still pending.
expect(pending).toHaveLength(2);
const runRow = pending.find((p) => p.id === fired.row_id);
expect(runRow?.recurrence).toBeNull(); // never re-armed into a phantom series
expect(new Date(runRow!.process_after).getTime()).toBeLessThanOrEqual(Date.now());
});
it('task object exposes origin_session_id and created_at', async () => {
const r = await dispatch(
{
id: 'ro',
command: 'tasks-create',
args: { prompt: 'x', name: 'o', process_after: '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(r.ok).toBe(true);
if (!r.ok) return;
const d = r.data as { origin_session_id: string | null; created_at: string };
expect(d.origin_session_id).toBe('chat-1'); // the session that created it
expect(d.created_at).toBeTruthy();
});
it('each task gets its own isolated session, and list fans out across them', async () => {
const a = await dispatch(
{ id: 'r-a', command: 'tasks-create', args: { prompt: 'task A', process_after: '2026-01-15T09:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
const b = await dispatch(
{ id: 'r-b', command: 'tasks-create', args: { prompt: 'task B', process_after: '2026-01-15T09:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(a.ok && b.ok).toBe(true);
if (!a.ok || !b.ok) return;
const ta = a.data as { session_id: string; series_id: string };
const tb = b.data as { session_id: string; series_id: string };
// Distinct per-series sessions — not one shared system:tasks session.
expect(ta.session_id).not.toBe(tb.session_id);
// list (no --session) fans out across every task session in the group.
const list = await dispatch({ id: 'r-l', command: 'tasks-list', args: {} }, agentCtx('ag-1', 'chat-1'));
expect(list.ok).toBe(true);
if (!list.ok) return;
const ids = (list.data as Array<{ series_id: string }>).map((t) => t.series_id);
expect(ids).toContain(ta.series_id);
expect(ids).toContain(tb.series_id);
});
it('list enriches each series with run history (CronJob view)', async () => {
const created = await dispatch(
{
id: 'r-agg',
command: 'tasks-create',
args: { prompt: 'brain digest', recurrence: '0 9 * * *', 'process-after': '2026-01-15T09:05:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { session_id, series_id } = created.data as { session_id: string; series_id: string };
// Seed three completed fires for this series into its own session inbound.db.
const db = new Database(inboundDbPath('ag-1', session_id));
const ins = db.prepare(
'INSERT INTO messages_in (id, seq, timestamp, status, tries, kind, content, series_id, process_after) ' +
"VALUES (?, ?, datetime('now'), 'completed', 0, 'task', '{}', ?, ?)",
);
ins.run('run-1', 100, series_id, '2026-01-15T09:02:00Z');
ins.run('run-2', 102, series_id, '2026-01-15T09:03:00Z');
ins.run('run-3', 104, series_id, '2026-01-15T09:04:00Z');
db.close();
const list = await dispatch({ id: 'r-agg-l', command: 'tasks-list', args: {} }, agentCtx('ag-1', 'chat-1'));
expect(list.ok).toBe(true);
if (!list.ok) return;
const row = (list.data as Array<Record<string, unknown>>).find((t) => t.series_id === series_id);
expect(row).toBeDefined();
expect(row?.runs).toBe(3);
expect(row?.last_run).toBe('2026-01-15T09:04:00Z'); // max completed process_after
expect(String(row?.next_run)).toMatch(/^2026-01-15T09:05:00/); // the live pending occurrence
expect(row?.schedule).toBe('0 9 * * *');
expect(row?.log).toBe(`tasks/${series_id}.md`);
});
// The schedule→wake primitive without a container: a task created through the
// real `ncl tasks create` path must land in the agent group's system session
// AND be counted by the same due-message query the host sweep uses to decide a
// wake. Goes red if trigger defaulting, system-session routing, or the due
// predicate ever drift apart.
describe('a due task makes the system session wakeable', () => {
it('countDueMessages sees a past task and ignores a future one', async () => {
const created = await dispatch(
{ id: 'r-due', command: 'tasks-create', args: { prompt: 'run me', 'process-after': '2020-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const systemId = (created.data as { session_id: string }).session_id;
const dueDb = new Database(inboundDbPath('ag-1', systemId), { readonly: true });
expect(countDueMessages(dueDb)).toBe(1); // host sweep would wake this session
dueDb.close();
// A far-future task in the same system session is not yet due.
const future = await dispatch(
{ id: 'r-fut', command: 'tasks-create', args: { prompt: 'later', 'process-after': '2999-01-01T00:00:00Z' } },
agentCtx('ag-1', 'chat-1'),
);
expect(future.ok).toBe(true);
const stillDb = new Database(inboundDbPath('ag-1', systemId), { readonly: true });
expect(countDueMessages(stillDb)).toBe(1); // still just the one past task
stillDb.close();
});
});
describe('append-log', () => {
const logFile = (folder: string, series: string) => `${TEST_DIR}/groups/${folder}/tasks/${series}.md`;
it('writes a host-timestamped line to the run log and creates the file (explicit --id)', async () => {
const resp = await dispatch(
{ id: 'al-1', command: 'tasks-append-log', args: { id: 'my-task-1', msg: 'did the thing; it worked' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
const content = fs.readFileSync(logFile('ag-1', 'my-task-1'), 'utf8').trim();
expect(content).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z — did the thing; it worked$/);
});
it('derives the series from the caller task session when --id is omitted', async () => {
const created = await dispatch(
{
id: 'al-c',
command: 'tasks-create',
args: { name: 'derive-me', prompt: 'x', 'process-after': '2999-01-01T00:00:00Z' },
},
agentCtx('ag-1', 'chat-1'),
);
expect(created.ok).toBe(true);
if (!created.ok) return;
const { series_id, session_id } = created.data as { series_id: string; session_id: string };
// The fire runs INSIDE that task session, so no --id is needed.
const resp = await dispatch(
{ id: 'al-2', command: 'tasks-append-log', args: { msg: 'auto-derived run' } },
agentCtx('ag-1', session_id),
);
expect(resp.ok).toBe(true);
if (!resp.ok) return;
expect((resp.data as { series: string }).series).toBe(series_id);
expect(fs.readFileSync(logFile('ag-1', series_id), 'utf8')).toContain('auto-derived run');
});
it('requires --msg', async () => {
const resp = await dispatch(
{ id: 'al-3', command: 'tasks-append-log', args: { id: 'my-task-1' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) expect(resp.error.message).toContain('--msg is required');
});
it('errors when there is no --id and the caller is not in a task session', async () => {
// chat-1 is a normal chat session, not system:tasks:* → nothing to derive.
const resp = await dispatch(
{ id: 'al-4', command: 'tasks-append-log', args: { msg: 'orphan' } },
agentCtx('ag-1', 'chat-1'),
);
expect(resp.ok).toBe(false);
if (!resp.ok) expect(resp.error.message).toMatch(/--id is required/);
});
});
});
describe('formatTasksTable', () => {
const now = Date.parse('2026-01-15T09:05:30Z');
const rows = [
{
series_id: 'task-5bbe082a-6298-4699',
schedule: '* * * * *',
runs: 7,
failed_runs: 2,
last_run: '2026-01-15T09:04:30Z',
next_run: '2026-01-15T09:06:00Z',
status: 'pending',
log: 'tasks/task-5bbe082a.md',
created_at: '2026-01-15T08:05:30Z', // 1h before now
prompt: 'You are NanoClaw, wired into the company brain, your job this run is to read it',
},
];
it('renders an aligned table with run history', () => {
const lines = formatTasksTable(rows, now).split('\n');
expect(lines[0]).toMatch(/SERIES\s+SCHEDULE\s+RUNS\s+FAILED\s+LAST RUN\s+NEXT RUN\s+STATUS\s+AGE\s+PROMPT/);
expect(lines[1]).toContain('1h'); // AGE column — created 1h ago
expect(lines[1]).toContain('task-5bbe082a-6298-4699'); // FULL series id — copy-pasteable into `tasks get --id`
expect(lines[1]).toContain('* * * * *');
expect(lines[1]).toContain('1m ago'); // 09:04:30 vs 09:05:30
expect(lines[1]).toContain('in 30s'); // 09:06:00 vs 09:05:30
expect(lines[1]).toContain('…'); // prompt truncated
});
it('handles a never-fired series and an empty list', () => {
expect(formatTasksTable([], now)).toBe('No tasks.');
const oneShot = formatTasksTable(
[
{
series_id: 'task-x',
schedule: 'once',
runs: 0,
last_run: null,
next_run: '2026-01-15T09:00:00Z',
status: 'pending',
},
],
now,
).split('\n')[1];
expect(oneShot).toContain('once');
expect(oneShot).toMatch(/\bdue\b/); // next_run in the past → due
expect(oneShot).toContain('-'); // last_run '-' (never fired)
});
});
describe('deep verb help (ncl tasks help create)', () => {
it('resolves through the dispatcher fallback and renders the full contract + examples', async () => {
// Side-effect import mirrors the CLI server boot: registers <plural>-help.
await import('../commands/index.js');
const resp = await dispatch({ id: 'h1', command: 'tasks-help-create', args: {} }, { caller: 'host' });
expect(resp.ok).toBe(true);
if (resp.ok) {
const text = resp.data as string;
expect(text).toContain('ncl tasks create');
expect(text).toContain('wakeAgent'); // full multi-line script contract present
expect(text).toContain('Examples:'); // examples block rendered
}
});
it('rejects an unknown verb with a pointer back to resource help', async () => {
await import('../commands/index.js');
const resp = await dispatch({ id: 'h2', command: 'tasks-help-frobnicate', args: {} }, { caller: 'host' });
expect(resp.ok).toBe(false);
});
});
+784
View File
@@ -0,0 +1,784 @@
import { randomUUID } from 'crypto';
import fs from 'fs';
import type Database from 'better-sqlite3';
import { CronExpressionParser } from 'cron-parser';
import { GROUPS_DIR, TIMEZONE } from '../../config.js';
import { getAgentGroup } from '../../db/agent-groups.js';
import {
findTaskSessions,
getActiveSessions,
getSession,
isTaskThread,
TASKS_SYSTEM_THREAD_ID,
} from '../../db/sessions.js';
import {
cancelAllTasks,
cancelTask,
deleteTask,
insertTaskRow,
pauseTask,
resumeTask,
updateTask,
type TaskUpdate,
} from '../../modules/scheduling/db.js';
import { inboundDbPath, resolveTaskSession, withInboundDb } from '../../session-manager.js';
import { parseZonedToUtc } from '../../timezone.js';
import { registerResource } from '../crud.js';
import { formatTasksTable } from '../format-tasks.js';
import type { CallerContext } from '../frame.js';
type TaskStatus = 'pending' | 'paused';
interface TaskRow {
row_id: string;
series_id: string | null;
status: string;
process_after: string | null;
recurrence: string | null;
content: string;
timestamp: string;
tries: number;
seq: number;
}
interface ScopedSession {
id: string;
agent_group_id: string;
}
function str(value: unknown): string | undefined {
return typeof value === 'string' && value.length > 0 ? value : undefined;
}
function bool(value: unknown): boolean {
return value === true || value === 'true' || value === '1';
}
/**
* Short, readable, filesystem/thread-safe task id. With a name `<slug>-<4hex>`
* (e.g. "Morning joke" `morning-joke-a25c`); without `t-<6hex>`. Always
* matches /^[a-z0-9-]+$/ so it is safe as a thread suffix (`system:tasks:<id>`),
* a filename (`tasks/<id>.md`), and a copy-pasteable --id.
*/
function makeTaskId(name: unknown): string {
const hex = (n: number): string => randomUUID().replace(/-/g, '').slice(0, n);
const slug =
typeof name === 'string'
? name
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.slice(0, 24)
.replace(/-+$/g, '')
: '';
return slug ? `${slug}-${hex(4)}` : `t-${hex(6)}`;
}
function parseProcessAfter(value: unknown): string {
const raw = str(value);
if (!raw) throw new Error('--process-after is required');
const date = parseZonedToUtc(raw, TIMEZONE);
if (Number.isNaN(date.getTime())) throw new Error(`invalid --process-after: ${raw}`);
return date.toISOString();
}
/**
* First-run timestamp for a new task. When a recurrence is given but no
* --process-after, derive the first fire from the cron grid (in TIMEZONE) so the
* common recurring case is a single flag `--recurrence "0 9 * * 1-5"` with no
* redundant, easily-stale hand-picked instant. --process-after is still required
* for one-shots (no recurrence to derive from) and still wins when supplied.
*/
function firstRunIso(value: unknown, recurrence: string | null): string {
if (str(value) === undefined && recurrence) {
const next = CronExpressionParser.parse(recurrence, { tz: TIMEZONE }).next().toISOString();
if (!next) throw new Error(`--recurrence has no upcoming run: ${recurrence}`);
return next;
}
return parseProcessAfter(value);
}
function normalizeNullableString(value: unknown): string | null | undefined {
if (value === undefined) return undefined;
if (value === null) return null;
if (typeof value !== 'string') return String(value);
const trimmed = value.trim();
if (trimmed === '' || trimmed === 'null' || trimmed === 'none') return null;
return value;
}
function validateRecurrence(value: string | null | undefined): void {
if (!value) return;
try {
CronExpressionParser.parse(value, { tz: TIMEZONE });
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
throw new Error(`invalid --recurrence: ${msg}`, { cause: err });
}
}
/**
* Frequency guard: refuse recurrences more frequent than 4 fires/day unless
* the agent explicitly overrides. Frequent tasks burn the user's quota (or
* get their account banned) the sanctioned pattern is a slower cron plus a
* pre-task gate script that checks an external condition and only wakes the
* agent when something changed (`ncl tasks create --help`). Counted over the
* next 24h from now in the instance timezone, so uneven crons are judged by
* what they would actually do.
*/
const MAX_DAILY_FIRES = 4;
const RECURRENCE_LIMIT_WARNING =
'Warning: this task has not been scheduled. Frequent running tasks consume the ' +
"user's subscription quota or unnecessarily use tokens and can cause the user's " +
'account to be banned. Instead, use a pre-task run script that you write that can ' +
'check some kind of external condition, usually via one or more API calls. The ' +
'script returns a decision programmatically whether the task needs to be run now ' +
'or not. For example, an API call to GitHub to check if there are open PRs, and ' +
'only run when there are new open PRs.\n' +
'Run `ncl tasks create --help` to get full directions on how to write a script and test it.\n\n' +
'Note: if and only if you explicitly need to schedule a task more frequently and ' +
"you've verified with the user that they understand and that this is what they " +
'want and based on your judgment you agree that this is the right thing to do in ' +
'this situation, you can override this with --dangerously-override-recurrence-limit';
function enforceRecurrenceLimit(recurrence: string | null, override: boolean, hasScript: boolean): void {
// A gate script IS the sanctioned mitigation the warning steers toward — a
// script-gated fire that finds nothing never wakes the agent, so scripted
// tasks may run at any cadence without the override.
if (!recurrence || override || hasScript) return;
const horizon = Date.now() + 24 * 60 * 60 * 1000;
const interval = CronExpressionParser.parse(recurrence, { tz: TIMEZONE });
let fires = 0;
while (fires <= MAX_DAILY_FIRES) {
const next = interval.next();
if (next.getTime() > horizon) break;
fires++;
}
if (fires > MAX_DAILY_FIRES) throw new Error(RECURRENCE_LIMIT_WARNING);
}
function statusFilter(args: Record<string, unknown>): TaskStatus | undefined {
const status = str(args.status);
if (!status) return undefined;
if (status !== 'pending' && status !== 'paused') {
throw new Error('--status must be pending or paused');
}
return status;
}
function groupArg(args: Record<string, unknown>, ctx: CallerContext): string | undefined {
if (ctx.caller === 'agent') return ctx.agentGroupId;
return str(args.group) ?? str(args.agent_group_id);
}
function ownSession(sessionId: string, ctx: CallerContext): ScopedSession {
const session = getSession(sessionId);
if (!session) throw new Error(`session not found: ${sessionId}`);
if (ctx.caller === 'agent' && session.agent_group_id !== ctx.agentGroupId) {
throw new Error(`session not found: ${sessionId}`);
}
return { id: session.id, agent_group_id: session.agent_group_id };
}
function selectedSessions(args: Record<string, unknown>, ctx: CallerContext): ScopedSession[] {
const sessionId = str(args.session);
if (sessionId) return [ownSession(sessionId, ctx)];
const group = groupArg(args, ctx);
if (group) {
// One session per live task series — the loops below already fan out across them.
return findTaskSessions(group).map((s) => ({ id: s.id, agent_group_id: s.agent_group_id }));
}
if (ctx.caller === 'agent') return [];
return getActiveSessions().map((s) => ({ id: s.id, agent_group_id: s.agent_group_id }));
}
function withInbound<T>(session: ScopedSession, fn: (db: Database.Database) => T): T | undefined {
if (!fs.existsSync(inboundDbPath(session.agent_group_id, session.id))) return undefined;
return withInboundDb(session.agent_group_id, session.id, fn);
}
function parseContent(raw: string): { prompt: string; script: string | null; originSessionId: string | null } {
try {
const parsed = JSON.parse(raw) as Record<string, unknown>;
return {
prompt: typeof parsed.prompt === 'string' ? parsed.prompt : '',
script: typeof parsed.script === 'string' ? parsed.script : null,
originSessionId: typeof parsed.originSessionId === 'string' ? parsed.originSessionId : null,
};
} catch {
// LEGACY-COMPAT(v1-tasks): plain-string content from rows that predate the
// JSON envelope. Removable once no pre-v2 session DBs remain in the wild.
return { prompt: raw, script: null, originSessionId: null };
}
}
function toOutput(session: ScopedSession, row: TaskRow) {
const content = parseContent(row.content);
return {
agent_group_id: session.agent_group_id,
session_id: session.id,
series_id: row.series_id ?? row.row_id,
row_id: row.row_id,
status: row.status,
process_after: row.process_after,
recurrence: row.recurrence,
prompt: content.prompt.length > 120 ? content.prompt.slice(0, 117) + '...' : content.prompt,
has_script: content.script ? 1 : 0,
origin_session_id: content.originSessionId, // which session created the task (null for CLI-created)
created_at: row.timestamp,
tries: row.tries,
};
}
function selectLiveTasks(db: Database.Database, status?: TaskStatus): TaskRow[] {
const statusSql = status ? 'status = ?' : "status IN ('pending', 'paused')";
return db
.prepare(
`SELECT id AS row_id, series_id, status, process_after, recurrence, content, timestamp, tries, MAX(seq) AS seq
FROM messages_in
WHERE kind = 'task'
AND ${statusSql}
GROUP BY series_id
ORDER BY datetime(process_after) ASC, seq ASC`,
)
.all(...(status ? [status] : [])) as TaskRow[];
}
function selectTask(db: Database.Database, id: string): TaskRow | undefined {
return db
.prepare(
`SELECT id AS row_id, series_id, status, process_after, recurrence, content, timestamp, tries, seq
FROM messages_in
WHERE kind = 'task'
AND (id = ? OR series_id = ?)
ORDER BY CASE WHEN status IN ('pending', 'paused') THEN 0 ELSE 1 END, seq DESC
LIMIT 1`,
)
.get(id, id) as TaskRow | undefined;
}
function taskId(args: Record<string, unknown>): string {
const id = str(args.id);
if (!id) throw new Error('task series id is required');
return id;
}
function createTask(args: Record<string, unknown>, ctx: CallerContext) {
const group = groupArg(args, ctx);
if (!group) throw new Error('--group is required');
const prompt = str(args.prompt);
if (!prompt) throw new Error('--prompt is required');
const recurrence = normalizeNullableString(args.recurrence) ?? null;
validateRecurrence(recurrence);
const script = normalizeNullableString(args.script) ?? null;
enforceRecurrenceLimit(recurrence, bool(args.dangerously_override_recurrence_limit), script != null);
const processAfter = firstRunIso(args.process_after, recurrence);
const id = makeTaskId(args.name);
const originSessionId = ctx.caller === 'agent' ? ctx.sessionId : null;
// Each series runs in its own isolated session; point the fire at its own log.
const { session } = resolveTaskSession(group, id);
const promptWithLog =
`${prompt}\n\n` +
`[A task serves the user two separate ways — do whichever the task above asks for, and ALWAYS the run log:\n` +
`• MESSAGE (only if asked): if the task says to report/notify the user, send your result with an EXPLICIT destination — <message to="name">…</message> or send_message({ to: "name", … }). This run has no chat attached: an unaddressed reply is DISCARDED, so the explicit send is the ONLY thing the user receives.\n` +
`• RUN LOG (ALWAYS — even if you sent no message and did nothing else this run): after any sends, end the run with:\n` +
` ncl tasks append-log --msg "<what you did, and why it mattered>"\n` +
` Write it like a work-log entry a human keeps — concrete: what you did and WHY (a no-op run still gets a line saying why nothing was needed). If you wrote or modified files this run, name them in --msg. Not a greeting, not a copy of the message you sent. The host stamps the UTC time (do NOT add one), do NOT edit tasks/${id}.md by hand, and this NEVER goes to the user.\n` +
`Need context from past runs? Read tasks/${id}.md first.]`;
const created = withInbound(session, (db) => {
insertTaskRow(db, {
id,
seriesId: id,
processAfter,
recurrence,
content: JSON.stringify({ prompt: promptWithLog, script, originSessionId }),
});
return selectTask(db, id);
});
if (!created) throw new Error('task system session inbound.db not found');
return toOutput(session, created);
}
/**
* Append one host-timestamped line to a task's run log
* (`<GROUPS_DIR>/<folder>/tasks/<series>.md`). This is NOT a delivery it writes
* nothing to messages_out; it just records what happened so the agent (and human)
* can see when and why each fire ran. Inside a task fire the series is derived from
* the caller's own task session, so the agent supplies only --msg.
*/
function appendTaskLog(
args: Record<string, unknown>,
ctx: CallerContext,
): { series: string; timestamp: string; path: string; ok: true } {
const msg = str(args.msg);
if (!msg) throw new Error('--msg is required');
let series = str(args.id);
let group = groupArg(args, ctx);
if (!series && ctx.caller === 'agent' && ctx.sessionId) {
const sess = getSession(ctx.sessionId);
if (sess && sess.thread_id && isTaskThread(sess.thread_id)) {
series = sess.thread_id.slice(`${TASKS_SYSTEM_THREAD_ID}:`.length);
group ??= sess.agent_group_id;
}
}
if (!series) throw new Error('--id is required (no task session to derive it from)');
// Charset guard is the security boundary here: blocks path traversal and keeps
// the id safe as a filename / thread suffix. Group scope is already enforced by
// groupArg (a cli_scope=group caller can only ever resolve its own folder), so a
// foreign id at worst writes a stray log under the caller's OWN folder — no leak.
if (!/^[a-z0-9-]+$/.test(series)) throw new Error(`invalid task id: ${series}`);
if (!group) throw new Error('could not resolve the agent group');
const ag = getAgentGroup(group);
if (!ag) throw new Error(`agent group not found: ${group}`);
const timestamp = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
const dir = `${GROUPS_DIR}/${ag.folder}/tasks`;
const file = `${dir}/${series}.md`;
fs.mkdirSync(dir, { recursive: true });
fs.appendFileSync(file, `${timestamp}${msg}\n`);
return { series, timestamp, path: file, ok: true };
}
/**
* Run history for one task series, aggregated over its occurrence rows: number
* of successful fires, the last fire time, and failed fires (a row reaches
* `failed` after MAX_TRIES on a stuck claim). Cancelled occurrences are
* `cancelled`, not `completed`, so they never inflate the run count.
*/
function seriesStats(
db: Database.Database,
seriesKey: string,
): { runs: number; last_run: string | null; failed_runs: number } {
return db
.prepare(
`SELECT
COUNT(*) FILTER (WHERE status = 'completed') AS runs,
MAX(process_after) FILTER (WHERE status = 'completed') AS last_run,
COUNT(*) FILTER (WHERE status = 'failed') AS failed_runs
FROM messages_in
WHERE kind = 'task' AND (id = ? OR series_id = ?)`,
)
.get(seriesKey, seriesKey) as { runs: number; last_run: string | null; failed_runs: number };
}
/** Last ~10 lines of a series' run log (`tasks/<series>.md`), newest last. */
function tailRunLog(agentGroupId: string, seriesKey: string, lines = 10): string[] {
const ag = getAgentGroup(agentGroupId);
if (!ag) return [];
const file = `${GROUPS_DIR}/${ag.folder}/tasks/${seriesKey}.md`;
if (!fs.existsSync(file)) return [];
return fs.readFileSync(file, 'utf8').trimEnd().split('\n').filter(Boolean).slice(-lines);
}
/**
* A task series is CronJob-like: the live (pending/paused) row is the next run,
* and the `completed` rows are its run history. Enrich each listed series with
* that history run count, failures, last fire, next fire, schedule, and a
* pointer to the agent's own run log so `tasks list` reads as a compact
* run-history table.
*/
function enrichListRow(db: Database.Database, base: ReturnType<typeof toOutput>) {
const seriesKey = base.series_id;
const stats = seriesStats(db, seriesKey);
return {
...base,
schedule: base.recurrence ?? 'once',
runs: stats.runs,
failed_runs: stats.failed_runs,
last_run: stats.last_run,
next_run: base.process_after,
log: `tasks/${seriesKey}.md`,
};
}
function listTasks(args: Record<string, unknown>, ctx: CallerContext) {
const status = statusFilter(args);
const rows = [];
for (const session of selectedSessions(args, ctx)) {
const sessionRows = withInbound(session, (db) =>
selectLiveTasks(db, status).map((row) => enrichListRow(db, toOutput(session, row))),
);
if (sessionRows) rows.push(...sessionRows);
}
return rows;
}
function getTask(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
for (const session of selectedSessions(args, ctx)) {
const found = withInbound(session, (db) => {
const row = selectTask(db, id);
if (!row) return undefined;
const seriesKey = row.series_id ?? row.row_id;
const stats = seriesStats(db, seriesKey);
const content = parseContent(row.content);
return {
...toOutput(session, row),
prompt: content.prompt,
script: content.script,
origin_session_id: content.originSessionId,
completed_runs: stats.runs,
failed_runs: stats.failed_runs,
recent_log: tailRunLog(session.agent_group_id, seriesKey),
};
});
if (found) return found;
}
throw new Error(`task not found: ${id}`);
}
function mutateTask(
args: Record<string, unknown>,
ctx: CallerContext,
fn: (db: Database.Database, id: string) => number,
) {
const id = taskId(args);
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, (db) => fn(db, id)) ?? 0;
}
if (touched === 0) throw new Error(`no live task matched: ${id}`);
return { series_id: id, touched };
}
function updateTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
const update: TaskUpdate = {};
if (typeof args.prompt === 'string') update.prompt = args.prompt;
if (args.process_after !== undefined) update.processAfter = parseProcessAfter(args.process_after);
const recurrence = normalizeNullableString(args.recurrence);
const script = normalizeNullableString(args.script);
if (recurrence !== undefined) {
validateRecurrence(recurrence);
// Effective script AFTER this update: the new value when provided
// (including an explicit clear), else whatever the task already has.
let scriptAfter: string | null = script !== undefined ? script : null;
if (script === undefined) {
for (const session of selectedSessions(args, ctx)) {
const row = withInbound(session, (db) => selectTask(db, id));
if (row) {
scriptAfter = parseContent(row.content).script;
break;
}
}
}
enforceRecurrenceLimit(recurrence, bool(args.dangerously_override_recurrence_limit), scriptAfter != null);
update.recurrence = recurrence;
}
if (script !== undefined) update.script = script;
const fields = Object.keys(update);
if (fields.length === 0) throw new Error('nothing to update');
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, (db) => updateTask(db, id, update)) ?? 0;
}
if (touched === 0) throw new Error(`no live task matched: ${id}`);
return { series_id: id, touched, fields };
}
function cancelTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
if (!bool(args.all)) {
return mutateTask(args, ctx, cancelTask);
}
let touched = 0;
for (const session of selectedSessions(args, ctx)) {
touched += withInbound(session, cancelAllTasks) ?? 0;
}
return { cancelled: touched };
}
/**
* `ncl tasks run <id>` fire a task on demand without disturbing its schedule.
* Inserts a fresh pending occurrence (same series, content, no recurrence) due
* now, which the next sweep delivers through the normal fire path. Unlike
* `update --process-after now`, it neither consumes a one-shot nor force-advances
* a recurring series' armed occurrence, so it is safe for testing a task.
*/
function runTaskCommand(args: Record<string, unknown>, ctx: CallerContext) {
const id = taskId(args);
for (const session of selectedSessions(args, ctx)) {
const fired = withInbound(session, (db) => {
const row = selectTask(db, id);
if (!row) return undefined;
const seriesKey = row.series_id ?? row.row_id;
const rowId = makeTaskId(`${seriesKey}-run`);
// recurrence=NULL is load-bearing: a run-now row must not be re-armed by
// handleRecurrence into a phantom series.
insertTaskRow(db, {
id: rowId,
seriesId: seriesKey,
processAfter: new Date().toISOString(),
recurrence: null,
content: row.content,
});
return { series_id: seriesKey, row_id: rowId, status: 'pending' };
});
if (fired) return fired;
}
throw new Error(`task not found: ${id}`);
}
registerResource({
name: 'task',
plural: 'tasks',
table: 'messages_in',
description:
'Scheduled task — prompt plus run time. Tasks run from the agent group system session and the agent chooses delivery destination at fire time.',
idColumn: 'series_id',
scopeField: 'agent_group_id',
columns: [
{ name: 'series_id', type: 'string', description: 'Stable task handle.', generated: true },
{ name: 'agent_group_id', type: 'string', description: 'Agent group that owns the task.' },
{ name: 'session_id', type: 'string', description: 'System session that runs the task.' },
{ name: 'status', type: 'string', description: 'Live state.', enum: ['pending', 'paused'] },
{
name: 'process_after',
type: 'string',
// Not flagged required: with --recurrence the first run is derived from the
// cron grid (firstRunIso). Required only for one-shots, enforced in the
// create handler — so the generic col.required validator must stay off here.
description:
'Next run time (ISO 8601 or naive local). Required for one-shots; with --recurrence the first run is derived from the cron grid.',
updatable: true,
},
{ name: 'recurrence', type: 'string', description: 'Optional cron expression.', updatable: true },
{ name: 'prompt', type: 'string', description: 'Task prompt.', required: true, updatable: true },
{ name: 'script', type: 'string', description: 'Optional pre-task bash script.', updatable: true },
],
operations: {},
customOperations: {
list: {
access: 'open',
description: 'List live tasks with per-series run history (schedule, runs, failures, next fire).',
args: [
{ name: 'status', type: 'string', description: 'Filter by live state.', enum: ['pending', 'paused'] },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
{
name: 'all',
type: 'boolean',
description: 'List across all groups (host default when no --group; accepted for explicitness).',
},
],
handler: async (args, ctx) => listTasks(args, ctx),
// Server-rendered run-history table (frame `human` field) — the container
// agent gets the same legible view as the host CLI without a Bun-side
// formatter copy.
formatHuman: (rows) => formatTasksTable(rows as Parameters<typeof formatTasksTable>[0]),
},
get: {
access: 'open',
description: 'Get a task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => getTask(args, ctx),
},
create: {
access: 'open',
description:
`Create a scheduled task (recurring or one-shot) in the agent group system session.\n\n` +
`Requires --prompt plus EITHER --recurrence (recurring; first run derived from the cron grid) OR --process-after (one-shot, ISO 8601 or naive local). Always pass --name for a readable id.\n\n` +
`--script contract (pre-task gate, runs BEFORE the agent wakes):\n` +
` bash, 30s timeout, 1MB output cap. Its LAST stdout line must be JSON:\n` +
` {"wakeAgent": <bool>, "data": {...}}\n` +
` wakeAgent=false marks the run handled without waking the agent (zero tokens);\n` +
` wakeAgent=true wakes the agent with data attached to the prompt.\n` +
` DO: print the JSON as the very last line, exit 0, keep data small (a summary, not a dump).\n` +
` DON'T: print anything after the JSON, prompt for input, or rely on state from previous runs.\n` +
` Always test with bash -c '<script>' before scheduling.\n` +
` Persist state between fires under the group workspace (e.g. a last-seen id file).\n` +
` Use good judgement on whether to share with the user the script (only if they are technical), a description of the script condition, or whether there's no need.\n\n` +
`Frequency limit: recurrences more frequent than ${MAX_DAILY_FIRES} fires/day are refused unless the task\n` +
`carries a --script gate (the script decides whether each fire needs you — a gated fire that\n` +
`finds nothing costs zero tokens) or you pass --dangerously-override-recurrence-limit after\n` +
`the user explicitly confirmed they want an ungated frequent task.\n\n` +
`Failure backoff: a script that ERRORS repeatedly backs the series off (2,4,8,…60 min between fires; each errored fire counts as a failed run); after 8 consecutive failures the series is auto-paused with a note in its run log — fix the script, then \`ncl tasks resume <id>\`. A deliberate wakeAgent=false is a normal run and never backs off. \`ncl tasks get <id>\` shows failed_runs and the run log.`,
args: [
{
name: 'name',
type: 'string',
description: 'Short descriptive name → readable task id (<slug>-<hex>). Without it, ids are t-<hex>.',
},
{ name: 'prompt', type: 'string', description: 'Task prompt the agent wakes to.', required: true },
{
name: 'recurrence',
type: 'string',
description:
'Cron expression (instance TZ). First run derives from the cron grid when --process-after is omitted.',
},
{
name: 'dangerously_override_recurrence_limit',
type: 'boolean',
description:
'Schedule more than 4 fires/day anyway. Only after the user explicitly confirmed they understand the quota/token cost and you agree it is right.',
},
{
name: 'process_after',
type: 'string',
description: 'First/next run time (ISO 8601 or naive local). Required for one-shots.',
},
{
name: 'script',
type: 'string',
description: 'Pre-task gate script (bash) — see the --script contract above.',
},
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
],
examples: [
`# Recurring — --recurrence alone is enough; the first run comes off the cron grid:\nncl tasks create --name "sales briefing" --prompt "Send the weekday sales briefing" --recurrence "0 9 * * 1-5"`,
`# One-shot — --process-after required (UTC, offset, or naive-local in the instance TZ):\nncl tasks create --name "ping" --prompt "Remind me to call Dana" --process-after "tomorrow 18:00"`,
`# Monitor — script gates the run; the agent wakes only when something matters:\nncl tasks create --name "alert watch" --recurrence "*/15 * * * *" \\\n --prompt "Investigate the alerts in the script data and notify me if serious" \\\n --script 'c=$(curl -sf https://example.com/api/alerts | jq length) || exit 0\necho "{\\"wakeAgent\\": $([ "$c" -gt 0 ] && echo true || echo false), \\"data\\": {\\"alerts\\": $c}}"'`,
],
handler: async (args, ctx) => createTask(args, ctx),
},
'append-log': {
access: 'open',
description:
'Append a one-line run summary to a task run log (tasks/<id>.md).\n\nThe host stamps the UTC timestamp; you supply --msg. This is a LOG ENTRY, not a message — it sends nothing to anyone. Inside a task fire --id is auto-derived from your session. If you wrote or modified files during the run, name them in --msg.',
examples: [
`# Inside a task fire (--id auto-derived) — the run's work-log line:\nncl tasks append-log --msg "posted the daily digest to slack; one feed returned 403, skipped"`,
],
args: [
{
name: 'msg',
type: 'string',
description:
'Your work-log entry: what you did and why it mattered (like a human work log). The host prepends the UTC timestamp; this is logged, never sent to the user.',
required: true,
},
{
name: 'id',
type: 'string',
description: 'Task series id. Auto-derived when called from inside a task fire; required otherwise.',
},
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
],
handler: async (args, ctx) => appendTaskLog(args, ctx),
},
update: {
access: 'open',
description: 'Update a live task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{ name: 'prompt', type: 'string', description: 'Replace the task prompt.' },
{ name: 'process_after', type: 'string', description: 'New next-run time (ISO 8601 or naive local).' },
{ name: 'recurrence', type: 'string', description: 'New cron expression; "null"/"none" clears it (one-shot).' },
{
name: 'dangerously_override_recurrence_limit',
type: 'boolean',
description:
'Schedule more than 4 fires/day anyway. Only after the user explicitly confirmed they understand the quota/token cost and you agree it is right.',
},
{ name: 'script', type: 'string', description: 'New pre-task script; "null"/"none" removes it.' },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => updateTaskCommand(args, ctx),
},
cancel: {
access: 'open',
description: 'Cancel a live task by series id, or use --all as a kill switch.',
args: [
{ name: 'id', type: 'string', description: 'Task series id (omit with --all).' },
{ name: 'all', type: 'boolean', description: 'Cancel every live task in scope — kill switch.' },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => cancelTaskCommand(args, ctx),
},
run: {
access: 'open',
description:
'Fire a task now without changing its schedule (queues an extra run due immediately). Safe for testing — unlike update --process-after now, it neither consumes a one-shot nor advances a recurring series.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => runTaskCommand(args, ctx),
},
pause: {
access: 'open',
description: 'Pause a pending task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, pauseTask),
},
resume: {
access: 'open',
description: 'Resume a paused task by series id.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, resumeTask),
},
delete: {
access: 'open',
description: 'Hard-delete a task series and its history.',
args: [
{ name: 'id', type: 'string', description: 'Task series id.', required: true },
{
name: 'group',
type: 'string',
description: 'Agent group id (host callers; auto-filled to your own group inside a container).',
},
{ name: 'session', type: 'string', description: 'Limit to one task session id.' },
],
handler: async (args, ctx) => mutateTask(args, ctx, deleteTask),
},
},
});
@@ -1,80 +0,0 @@
/**
* Upgrade-path test for migration 019 (holds-approver-rule): in-flight
* pending_approvals rows created by the pre-contract code must come out with
* the approver rule the old click-auth gave them, and the sender table must be
* gone.
*/
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { closeDb, initTestDb, runMigrations } from './index.js';
import { migrations } from './migrations/index.js';
import type Database from 'better-sqlite3';
let db: Database.Database;
beforeEach(() => {
db = initTestDb();
// Everything up to — but not including — the holds-approver-rule migration.
runMigrations(
db,
migrations.filter((m) => m.name !== 'holds-approver-rule'),
);
});
afterEach(() => {
closeDb();
});
function hasTable(name: string): boolean {
return db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(name) !== undefined;
}
describe('migration 019 — holds-approver-rule', () => {
it('backfills approver_rule and agent_group_id on in-flight rows and drops the sender table', () => {
const now = new Date().toISOString();
db.prepare("INSERT INTO agent_groups (id, name, folder, created_at) VALUES ('ag-1', 'One', 'one', ?)").run(now);
db.prepare(
"INSERT INTO sessions (id, agent_group_id, messaging_group_id, thread_id, created_at) VALUES ('sess-1', 'ag-1', NULL, NULL, ?)",
).run(now);
// Legacy a2a hold: named approver ⇒ was exclusive.
db.prepare(
`INSERT INTO pending_approvals (approval_id, session_id, request_id, action, payload, created_at, approver_user_id, title, options_json)
VALUES ('appr-a2a', 'sess-1', 'appr-a2a', 'a2a_message_gate', '{}', ?, 'tg:dana', '', '[]')`,
).run(now);
// Legacy cli hold: no approver, no agent_group_id — click-auth fell back
// to the session's group; the backfill makes that anchoring explicit.
db.prepare(
`INSERT INTO pending_approvals (approval_id, session_id, request_id, action, payload, created_at, title, options_json)
VALUES ('appr-cli', 'sess-1', 'appr-cli', 'cli_command', '{}', ?, '', '[]')`,
).run(now);
// Legacy OneCLI hold: sessionless, agent_group_id already stamped.
db.prepare(
`INSERT INTO pending_approvals (approval_id, session_id, request_id, action, payload, created_at, agent_group_id, title, options_json)
VALUES ('oa-1', NULL, 'req-uuid', 'onecli_credential', '{}', ?, 'ag-1', '', '[]')`,
).run(now);
expect(hasTable('pending_sender_approvals')).toBe(true);
runMigrations(db); // applies only holds-approver-rule
const rows = db
.prepare('SELECT approval_id, approver_rule, approver_scope, agent_group_id FROM pending_approvals')
.all() as Array<{ approval_id: string; approver_rule: string; approver_scope: string; agent_group_id: string }>;
const byId = Object.fromEntries(rows.map((r) => [r.approval_id, r]));
expect(byId['appr-a2a']).toMatchObject({
approver_rule: 'exclusive',
approver_scope: 'group',
agent_group_id: 'ag-1',
});
expect(byId['appr-cli']).toMatchObject({
approver_rule: 'admins-of-scope',
approver_scope: 'group',
agent_group_id: 'ag-1',
});
expect(byId['oa-1']).toMatchObject({ approver_rule: 'admins-of-scope', agent_group_id: 'ag-1' });
expect(hasTable('pending_sender_approvals')).toBe(false);
});
});
@@ -1,44 +0,0 @@
import type { Migration } from './index.js';
/**
* The hold-record contract lands on `pending_approvals` (guarded-actions
* phase 1 see the guarded-actions decisions doc, decision 5):
*
* - `approver_rule` who may resolve the hold: 'exclusive' (only
* `approver_user_id`, e.g. an a2a policy's named approver) or
* 'admins-of-scope' (the admin chain of `agent_group_id`, plus the
* specific user the card was delivered to when `approver_user_id` is
* stamped the sender/channel "named-or-admin" semantic).
* - `approver_scope` the action's blast radius: 'global' holds (e.g.
* `roles grant`) can only be resolved by an owner or global admin; a
* scoped admin's click is rejected.
* - `dedup_key` in-flight dedup: while a pending row carries a key, a
* second request with the same key is dropped (replaces the sender
* table's UNIQUE(messaging_group_id, sender_identity)).
*
* Backfills: rows with a named approver were exclusive before this column
* existed; `agent_group_id` is stamped from the requesting session so
* click-auth no longer needs the session fallback.
*
* `pending_sender_approvals` is dropped: sender admission now holds through
* the approvals primitive (action 'sender_admit'). In-flight sender cards at
* upgrade time die with the table they are transient courtesy cards, and a
* new message from the same sender re-triggers one.
*/
export const migration019: Migration = {
version: 19,
name: 'holds-approver-rule',
up(db) {
db.exec(`ALTER TABLE pending_approvals ADD COLUMN approver_rule TEXT NOT NULL DEFAULT 'admins-of-scope';`);
db.exec(`ALTER TABLE pending_approvals ADD COLUMN approver_scope TEXT NOT NULL DEFAULT 'group';`);
db.exec(`ALTER TABLE pending_approvals ADD COLUMN dedup_key TEXT;`);
db.exec(`UPDATE pending_approvals SET approver_rule = 'exclusive' WHERE approver_user_id IS NOT NULL;`);
db.exec(
`UPDATE pending_approvals
SET agent_group_id = (SELECT s.agent_group_id FROM sessions s WHERE s.id = pending_approvals.session_id)
WHERE agent_group_id IS NULL AND session_id IS NOT NULL;`,
);
db.exec(`DROP INDEX IF EXISTS idx_pending_sender_approvals_mg;`);
db.exec(`DROP TABLE IF EXISTS pending_sender_approvals;`);
},
};
-2
View File
@@ -17,7 +17,6 @@ import { migration016 } from './016-messaging-group-instance.js';
import { moduleApprovalsPendingApprovals } from './module-approvals-pending-approvals.js';
import { moduleApprovalsTitleOptions } from './module-approvals-title-options.js';
import { migration018 } from './018-approvals-approver-user-id.js';
import { migration019 } from './019-holds-approver-rule.js';
export interface Migration {
version: number;
@@ -51,7 +50,6 @@ export const migrations: Migration[] = [
migration014,
migration015,
migration016,
migration019,
];
/** Row shape of PRAGMA foreign_key_check. Child rowids are stable across a
+15
View File
@@ -133,6 +133,21 @@ CREATE TABLE pending_questions (
created_at TEXT NOT NULL
);
-- Pending approvals for unknown senders (unknown_sender_policy='request_approval').
-- In-flight dedup via UNIQUE(messaging_group_id, sender_identity): a second
-- message from the same unknown sender while a card is pending is silently
-- dropped instead of spamming the admin.
CREATE TABLE pending_sender_approvals (
id TEXT PRIMARY KEY,
messaging_group_id TEXT NOT NULL REFERENCES messaging_groups(id),
agent_group_id TEXT NOT NULL REFERENCES agent_groups(id),
sender_identity TEXT NOT NULL, -- namespaced user id (channel_type:handle)
sender_name TEXT,
original_message TEXT NOT NULL, -- JSON of the original InboundEvent
approver_user_id TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(messaging_group_id, sender_identity)
);
`;
/**
+64 -1
View File
@@ -10,7 +10,7 @@ import fs from 'fs';
import path from 'path';
import { describe, it, expect, afterEach } from 'vitest';
import { getInboundSourceSessionId, migrateMessagesInTable } from './session-db.js';
import { ensureSchema, getInboundSourceSessionId, migrateMessagesInTable, syncProcessingAcks } from './session-db.js';
const TEST_DIR = '/tmp/nanoclaw-session-db-test';
const DB_PATH = path.join(TEST_DIR, 'inbound.db');
@@ -92,3 +92,66 @@ describe('migrateMessagesInTable', () => {
db.close();
});
});
describe('syncProcessingAcks — script-skip counter', () => {
function freshPair() {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
fs.mkdirSync(TEST_DIR, { recursive: true });
ensureSchema(DB_PATH, 'inbound');
const outPath = path.join(TEST_DIR, 'outbound.db');
ensureSchema(outPath, 'outbound');
return { inDb: new Database(DB_PATH), outDb: new Database(outPath) };
}
function seedTask(inDb: InstanceType<typeof Database>, id: string, content: Record<string, unknown>) {
inDb
.prepare(
`INSERT INTO messages_in (id, seq, timestamp, status, tries, kind, content, series_id)
VALUES (?, 2, datetime('now'), 'processing', 0, 'task', ?, ?)`,
)
.run(id, JSON.stringify(content), id);
}
function ack(outDb: InstanceType<typeof Database>, id: string, status: string) {
outDb
.prepare(
"INSERT OR REPLACE INTO processing_ack (message_id, status, status_changed) VALUES (?, ?, datetime('now'))",
)
.run(id, status);
}
const status = (inDb: InstanceType<typeof Database>, id: string) =>
(inDb.prepare('SELECT status FROM messages_in WHERE id = ?').get(id) as { status: string }).status;
it('script-skip:error ack lands the row as a FAILED run (streak-derivable history)', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'script-skip:error');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('failed');
});
it('a settled row is terminal — a lingering ack cannot flip failed back to completed', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'script-skip:error');
syncProcessingAcks(inDb, outDb);
ack(outDb, 't1', 'completed');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('failed');
});
it('plain completed ack completes the row as before', () => {
const { inDb, outDb } = freshPair();
seedTask(inDb, 't1', { prompt: 'p', script: 'x' });
ack(outDb, 't1', 'completed');
syncProcessingAcks(inDb, outDb);
expect(status(inDb, 't1')).toBe('completed');
});
});
+20 -8
View File
@@ -168,15 +168,25 @@ export function getMessageForRetry(
export function syncProcessingAcks(inDb: Database.Database, outDb: Database.Database): void {
const completed = outDb
.prepare("SELECT message_id FROM processing_ack WHERE status IN ('completed', 'failed')")
.all() as Array<{ message_id: string }>;
.prepare(
"SELECT message_id, status FROM processing_ack WHERE status IN ('completed', 'failed', 'script-skip:error')",
)
.all() as Array<{ message_id: string; status: string }>;
if (completed.length === 0) return;
const updateStmt = inDb.prepare("UPDATE messages_in SET status = 'completed' WHERE id = ? AND status != 'completed'");
// `script-skip:error` (pre-task script crashed) lands as a FAILED run —
// semantically true, and it lets recurrence derive the trailing failed
// streak from the occurrence rows themselves (no stored counter).
const completeStmt = inDb.prepare(
"UPDATE messages_in SET status = 'completed' WHERE id = ? AND status NOT IN ('completed', 'failed')",
);
const failStmt = inDb.prepare(
"UPDATE messages_in SET status = 'failed' WHERE id = ? AND status NOT IN ('completed', 'failed')",
);
inDb.transaction(() => {
for (const { message_id } of completed) {
updateStmt.run(message_id);
for (const { message_id, status } of completed) {
(status === 'script-skip:error' ? failStmt : completeStmt).run(message_id);
}
})();
}
@@ -294,9 +304,11 @@ export function migrateDeliveredTable(db: Database.Database): void {
}
}
// Adds columns added to messages_in after the initial v2 schema to
// pre-existing session DBs. No-op on fresh installs where the columns are
// in the baseline schema. Backfills existing rows so invariants hold.
// LEGACY-COMPAT(v1-tasks): adds columns added to messages_in after the initial
// v2 schema to pre-existing session DBs — this lazy, on-open migration IS the
// upgrade path for old installs (there is no central migration for session
// DBs). No-op on fresh installs where the columns are in the baseline schema.
// Backfills existing rows so invariants hold (series_id = id).
export function migrateMessagesInTable(db: Database.Database): void {
const cols = new Set(
(db.prepare("PRAGMA table_info('messages_in')").all() as Array<{ name: string }>).map((c) => c.name),
+59 -16
View File
@@ -3,6 +3,8 @@ import { getDb, hasTable } from './connection.js';
// ── Sessions ──
export const TASKS_SYSTEM_THREAD_ID = 'system:tasks';
export function createSession(session: Session): void {
getDb()
.prepare(
@@ -55,7 +57,14 @@ export function findSessionForAgent(
/** Find an active session scoped to an agent group (ignoring messaging group). */
export function findSessionByAgentGroup(agentGroupId: string): Session | undefined {
return getDb()
.prepare("SELECT * FROM sessions WHERE agent_group_id = ? AND status = 'active' ORDER BY created_at DESC LIMIT 1")
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND status = 'active'
AND NOT (messaging_group_id IS NULL AND thread_id IS NOT NULL AND thread_id LIKE 'system:%')
ORDER BY created_at DESC
LIMIT 1`,
)
.get(agentGroupId) as Session | undefined;
}
@@ -63,6 +72,44 @@ export function getSessionsByAgentGroup(agentGroupId: string): Session[] {
return getDb().prepare('SELECT * FROM sessions WHERE agent_group_id = ?').all(agentGroupId) as Session[];
}
export function findSystemSession(agentGroupId: string, threadId: string): Session | undefined {
return getDb()
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND messaging_group_id IS NULL
AND thread_id = ?
AND status = 'active'
ORDER BY created_at DESC
LIMIT 1`,
)
.get(agentGroupId, threadId) as Session | undefined;
}
/** Per-task session thread id for a scheduled task series. */
export function taskThreadId(seriesId: string): string {
return `${TASKS_SYSTEM_THREAD_ID}:${seriesId}`;
}
/** True for any task session thread — a per-series one or the legacy shared one. */
export function isTaskThread(threadId: string | null): boolean {
return threadId === TASKS_SYSTEM_THREAD_ID || (threadId?.startsWith(`${TASKS_SYSTEM_THREAD_ID}:`) ?? false);
}
/** All active task sessions for a group — one per live series, plus any legacy shared one. */
export function findTaskSessions(agentGroupId: string): Session[] {
return getDb()
.prepare(
`SELECT * FROM sessions
WHERE agent_group_id = ?
AND messaging_group_id IS NULL
AND status = 'active'
AND (thread_id = ? OR thread_id LIKE ?)
ORDER BY created_at DESC`,
)
.all(agentGroupId, TASKS_SYSTEM_THREAD_ID, `${TASKS_SYSTEM_THREAD_ID}:%`) as Session[];
}
export function getActiveSessions(): Session[] {
return getDb().prepare("SELECT * FROM sessions WHERE status = 'active'").all() as Session[];
}
@@ -155,11 +202,11 @@ export function createPendingApproval(
`INSERT OR IGNORE INTO pending_approvals
(approval_id, session_id, request_id, action, payload, created_at,
agent_group_id, channel_type, platform_id, platform_message_id, expires_at, status,
title, options_json, approver_user_id, approver_rule, approver_scope, dedup_key)
title, options_json, approver_user_id)
VALUES
(@approval_id, @session_id, @request_id, @action, @payload, @created_at,
@agent_group_id, @channel_type, @platform_id, @platform_message_id, @expires_at, @status,
@title, @options_json, @approver_user_id, @approver_rule, @approver_scope, @dedup_key)`,
@title, @options_json, @approver_user_id)`,
)
.run({
session_id: null,
@@ -170,21 +217,11 @@ export function createPendingApproval(
expires_at: null,
status: 'pending',
approver_user_id: null,
approver_rule: 'admins-of-scope',
approver_scope: 'group',
dedup_key: null,
...pa,
});
return result.changes > 0;
}
/** In-flight lookup for `requestApproval`'s dedup: any live row with this key blocks a repeat request. */
export function getPendingApprovalByDedupKey(dedupKey: string): PendingApproval | undefined {
return getDb().prepare('SELECT * FROM pending_approvals WHERE dedup_key = ? LIMIT 1').get(dedupKey) as
| PendingApproval
| undefined;
}
export function getPendingApproval(approvalId: string): PendingApproval | undefined {
return getDb().prepare('SELECT * FROM pending_approvals WHERE approval_id = ?').get(approvalId) as
| PendingApproval
@@ -240,9 +277,8 @@ export function getAskQuestionRender(
| undefined;
if (a?.title) return { title: a.title, options: JSON.parse(a.options_json) };
// Channel-registration approvals persist title/options_json the same way
// pending_approvals does — just SELECT and return. (Unknown-sender approvals
// are pending_approvals rows since the sender fold.)
// Channel-registration + unknown-sender approvals persist title/options_json
// the same way pending_approvals does — just SELECT and return.
if (hasTable(getDb(), 'pending_channel_approvals')) {
const c = getDb()
.prepare('SELECT title, options_json FROM pending_channel_approvals WHERE messaging_group_id = ?')
@@ -250,5 +286,12 @@ export function getAskQuestionRender(
if (c?.title) return { title: c.title, options: JSON.parse(c.options_json) };
}
if (hasTable(getDb(), 'pending_sender_approvals')) {
const s = getDb().prepare('SELECT title, options_json FROM pending_sender_approvals WHERE id = ?').get(id) as
| { title: string; options_json: string }
| undefined;
if (s?.title) return { title: s.title, options: JSON.parse(s.options_json) };
}
return undefined;
}
+1 -1
View File
@@ -254,7 +254,7 @@ async function deliverMessage(
const content = JSON.parse(msg.content);
// System actions — handle internally (schedule_task, cancel_task, etc.)
// System actions — handle internally (cli_request, etc.)
if (msg.kind === 'system') {
await handleSystemAction(content, session, inDb);
return;
+20
View File
@@ -13,6 +13,7 @@ import {
_resetStuckProcessingRowsForTesting,
decideStuckAction,
parseSqliteUtc,
shouldCloseTaskSession,
} from './host-sweep.js';
import type { Session } from './types.js';
@@ -334,3 +335,22 @@ describe('parseSqliteUtc', () => {
expect(parseSqliteUtc(bare)).toBe(Date.parse(bare + 'Z'));
});
});
describe('shouldCloseTaskSession', () => {
it('closes a spent per-task session (no live tasks, no container)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', false, 0)).toBe(true);
});
it('keeps it while a task is still live (recurring re-armed, or pending/paused)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', false, 1)).toBe(false);
});
it('keeps it while its container is running (mid-fire)', () => {
expect(shouldCloseTaskSession('system:tasks:task-1', true, 0)).toBe(false);
});
it('never touches non-task sessions', () => {
expect(shouldCloseTaskSession('telegram:12345', false, 0)).toBe(false);
expect(shouldCloseTaskSession(null, false, 0)).toBe(false);
});
});
+28 -1
View File
@@ -30,7 +30,7 @@ import type Database from 'better-sqlite3';
import fs from 'fs';
import { ensureEgressNetwork } from './egress-lockdown.js';
import { getActiveSessions } from './db/sessions.js';
import { getActiveSessions, isTaskThread, updateSession } from './db/sessions.js';
import { getAgentGroup } from './db/agent-groups.js';
import {
countDueMessages,
@@ -167,6 +167,15 @@ async function sweep(): Promise<void> {
setTimeout(sweep, SWEEP_INTERVAL_MS);
}
/** A per-task session with no live tasks and no running container is spent → close it. */
export function shouldCloseTaskSession(
threadId: string | null,
containerRunning: boolean,
liveTaskCount: number,
): boolean {
return isTaskThread(threadId) && !containerRunning && liveTaskCount === 0;
}
async function sweepSession(session: Session): Promise<void> {
const agentGroup = getAgentGroup(session.agent_group_id);
if (!agentGroup) return;
@@ -234,6 +243,24 @@ async function sweepSession(session: Session): Promise<void> {
const { handleRecurrence } = await import('./modules/scheduling/recurrence.js');
await handleRecurrence(inDb, session);
// MODULE-HOOK:scheduling-recurrence:end
// 6. GC spent task sessions. An isolated per-task session with no live task
// rows left (one-shot fired, or all cancelled/deleted) and no container
// running is dead — close it so it stops being swept and listed. Runs after
// recurrence so a just-fired recurring series has already re-armed its next
// pending row and is never collected. The per-task log file in the workspace
// is the durable history and survives the close.
if (isTaskThread(session.thread_id)) {
const liveTasks = (
inDb
.prepare("SELECT COUNT(*) AS c FROM messages_in WHERE kind = 'task' AND status IN ('pending', 'paused')")
.get() as { c: number }
).c;
if (shouldCloseTaskSession(session.thread_id, isContainerRunning(session.id), liveTasks)) {
updateSession(session.id, { status: 'closed' });
log.info('Closed spent task session', { sessionId: session.id, threadId: session.thread_id });
}
}
} finally {
inDb.close();
outDb?.close();
@@ -94,10 +94,6 @@ export async function handleCreateAgent(content: Record<string, unknown>, sessio
* a confined (non-global) agent group. `session` is the requesting parent.
*/
export const applyCreateAgent: ApprovalHandler = async ({ session, payload, notify }) => {
if (!session) {
log.warn('create_agent approval resolved without a session — dropping');
return;
}
const name = typeof payload.name === 'string' ? payload.name : '';
const instructions = typeof payload.instructions === 'string' ? payload.instructions : null;
@@ -4,10 +4,6 @@ import type { ApprovalHandler } from '../approvals/index.js';
import { performAgentRoute, type RoutableAgentMessage } from './agent-route.js';
export const applyA2aMessageGate: ApprovalHandler = async ({ session, payload, notify }) => {
if (!session) {
log.warn('a2a_message_gate approval resolved without a session — dropping');
return;
}
const { id, platform_id, content, in_reply_to } = payload;
if (typeof platform_id !== 'string' || !platform_id) {
notify('Message approved but the target agent group was missing from the request.');
@@ -13,21 +13,12 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { initTestDb, closeDb, runMigrations } from '../../db/index.js';
import { createAgentGroup } from '../../db/agent-groups.js';
import { getDb } from '../../db/connection.js';
import { createMessagingGroup } from '../../db/messaging-groups.js';
import { createSession, createPendingApproval, getPendingApproval, getSession } from '../../db/sessions.js';
import { createSession, createPendingApproval } from '../../db/sessions.js';
import { upsertUser } from '../permissions/db/users.js';
import { grantRole } from '../permissions/db/user-roles.js';
import { initSessionFolder } from '../../session-manager.js';
import { handleApprovalsResponse } from './response-handler.js';
import {
registerApprovalHandler,
registerApprovalRequestedHandler,
registerApprovalResolvedHandler,
requestApproval,
type ApprovalRequestedEvent,
type ApprovalResolvedEvent,
} from './primitive.js';
import { registerApprovalHandler, registerApprovalResolvedHandler, type ApprovalResolvedEvent } from './primitive.js';
vi.mock('../../container-runner.js', () => ({
wakeContainer: vi.fn().mockResolvedValue(undefined),
@@ -109,7 +100,7 @@ describe('approval-resolved callbacks', () => {
expect(events[0].outcome).toBe('reject');
expect(events[0].approval.approval_id).toBe('appr-reject-1');
expect(events[0].approval.action).toBe('test_reject_action');
expect(events[0].session?.id).toBe('sess-1');
expect(events[0].session.id).toBe('sess-1');
expect(events[0].userId).toBe('slack:admin-1');
});
@@ -159,43 +150,3 @@ describe('approval-resolved callbacks', () => {
expect(events).toEqual(['boom', 'after']);
});
});
describe('approval-requested callbacks', () => {
it('fires when requestApproval creates a hold, with the row and the delivered approver', async () => {
// The approver needs a reachable DM for the delivery walk.
createMessagingGroup({
id: 'mg-dm-admin',
channel_type: 'slack',
platform_id: 'dm-admin',
name: 'Admin DM',
is_group: 0,
unknown_sender_policy: 'public',
created_at: now(),
});
getDb()
.prepare(`INSERT INTO user_dms (user_id, channel_type, messaging_group_id, resolved_at) VALUES (?, ?, ?, ?)`)
.run('slack:admin-1', 'slack', 'mg-dm-admin', now());
const events: ApprovalRequestedEvent[] = [];
registerApprovalRequestedHandler((event) => {
if (event.approval.action === 'test_requested_action') events.push(event);
});
await requestApproval({
session: getSession('sess-1')!,
agentName: 'Agent',
action: 'test_requested_action',
payload: { thing: 1 },
title: 'Test hold',
question: 'Approve the thing?',
});
expect(events).toHaveLength(1);
expect(events[0].deliveredTo).toBe('slack:admin-1');
expect(events[0].session?.id).toBe('sess-1');
expect(events[0].approval.agent_group_id).toBe('ag-1');
expect(events[0].approval.approver_rule).toBe('admins-of-scope');
// The event carries the live row.
expect(getPendingApproval(events[0].approval.approval_id)).toBeDefined();
});
});
-222
View File
@@ -1,222 +0,0 @@
/**
* mayResolve matrix the one click-authorization rule for every hold.
*
* Covers each approver-rule kind × clicker role × approver scope, including:
* - exclusive named approvers (a2a policy semantics: nobody else, not even
* an owner, may resolve)
* - admins-of-scope with and without a delivered approver (the
* sender/channel "named-or-admin" semantic)
* - the null-anchor variant (owners + global admins only)
* - the D1 fix: a 'global'-scope hold rejects a scoped admin's click even
* though the approver rule would otherwise accept it
*
* Plus an end-to-end D1 regression through the real response handler: a
* global-blast CLI hold (e.g. roles grant) clicked by a scoped admin is
* ignored; the owner's click resolves it.
*/
import * as fs from 'fs';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { initTestDb, closeDb, runMigrations } from '../../db/index.js';
import { createAgentGroup } from '../../db/agent-groups.js';
import { createSession, createPendingApproval, getPendingApproval } from '../../db/sessions.js';
import { upsertUser } from '../permissions/db/users.js';
import { grantRole } from '../permissions/db/user-roles.js';
import { initSessionFolder } from '../../session-manager.js';
import { approverRuleOf, mayResolve } from './approver-rule.js';
import { registerApprovalHandler } from './primitive.js';
import { handleApprovalsResponse } from './response-handler.js';
vi.mock('../../container-runner.js', () => ({
wakeContainer: vi.fn().mockResolvedValue(undefined),
}));
vi.mock('../../config.js', async () => {
const actual = await vi.importActual('../../config.js');
return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-approver-rule' };
});
const TEST_DIR = '/tmp/nanoclaw-test-approver-rule';
function now() {
return new Date().toISOString();
}
const OWNER = 'slack:owner';
const GLOBAL_ADMIN = 'slack:global-admin';
const SCOPED_ADMIN = 'slack:scoped-admin'; // admin @ ag-1
const OTHER_ADMIN = 'slack:other-admin'; // admin @ ag-2
const DELIVEREE = 'slack:deliveree'; // no role — the user a card was delivered to
const RANDO = 'slack:rando'; // no role
beforeEach(() => {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true, force: true });
fs.mkdirSync(TEST_DIR, { recursive: true });
const db = initTestDb();
runMigrations(db);
createAgentGroup({ id: 'ag-1', name: 'One', folder: 'one', agent_provider: null, created_at: now() });
createAgentGroup({ id: 'ag-2', name: 'Two', folder: 'two', agent_provider: null, created_at: now() });
for (const id of [OWNER, GLOBAL_ADMIN, SCOPED_ADMIN, OTHER_ADMIN, DELIVEREE, RANDO]) {
upsertUser({ id, kind: 'slack', display_name: id, created_at: now() });
}
grantRole({ user_id: OWNER, role: 'owner', agent_group_id: null, granted_by: null, granted_at: now() });
grantRole({ user_id: GLOBAL_ADMIN, role: 'admin', agent_group_id: null, granted_by: null, granted_at: now() });
grantRole({ user_id: SCOPED_ADMIN, role: 'admin', agent_group_id: 'ag-1', granted_by: null, granted_at: now() });
grantRole({ user_id: OTHER_ADMIN, role: 'admin', agent_group_id: 'ag-2', granted_by: null, granted_at: now() });
});
afterEach(() => {
closeDb();
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true, force: true });
});
describe('mayResolve matrix', () => {
it('exclusive: only the named user, regardless of rank', () => {
const e = { kind: 'exclusive', approverUserId: DELIVEREE } as const;
expect(mayResolve(e, 'group', DELIVEREE)).toBe(true);
expect(mayResolve(e, 'group', OWNER)).toBe(false);
expect(mayResolve(e, 'group', GLOBAL_ADMIN)).toBe(false);
expect(mayResolve(e, 'group', SCOPED_ADMIN)).toBe(false);
expect(mayResolve(e, 'group', RANDO)).toBe(false);
expect(mayResolve(e, 'group', null)).toBe(false);
});
it('exclusive ∩ global scope: the named user must also be owner/global admin', () => {
expect(mayResolve({ kind: 'exclusive', approverUserId: DELIVEREE }, 'global', DELIVEREE)).toBe(false);
expect(mayResolve({ kind: 'exclusive', approverUserId: OWNER }, 'global', OWNER)).toBe(true);
expect(mayResolve({ kind: 'exclusive', approverUserId: GLOBAL_ADMIN }, 'global', GLOBAL_ADMIN)).toBe(true);
});
it('admins-of-scope(group) with a delivered approver: named-or-admin', () => {
const e = { kind: 'admins-of-scope', agentGroupId: 'ag-1', deliveredTo: DELIVEREE } as const;
expect(mayResolve(e, 'group', DELIVEREE)).toBe(true); // delivered-to shortcut
expect(mayResolve(e, 'group', SCOPED_ADMIN)).toBe(true);
expect(mayResolve(e, 'group', GLOBAL_ADMIN)).toBe(true);
expect(mayResolve(e, 'group', OWNER)).toBe(true);
expect(mayResolve(e, 'group', OTHER_ADMIN)).toBe(false); // admin of another group
expect(mayResolve(e, 'group', RANDO)).toBe(false);
expect(mayResolve(e, 'group', null)).toBe(false);
});
it('admins-of-scope(group) without a delivered approver: pure admin chain', () => {
const e = { kind: 'admins-of-scope', agentGroupId: 'ag-1', deliveredTo: null } as const;
expect(mayResolve(e, 'group', SCOPED_ADMIN)).toBe(true);
expect(mayResolve(e, 'group', GLOBAL_ADMIN)).toBe(true);
expect(mayResolve(e, 'group', OWNER)).toBe(true);
expect(mayResolve(e, 'group', DELIVEREE)).toBe(false);
expect(mayResolve(e, 'group', OTHER_ADMIN)).toBe(false);
});
it('admins-of-scope(null): owners and global admins only', () => {
const e = { kind: 'admins-of-scope', agentGroupId: null, deliveredTo: null } as const;
expect(mayResolve(e, 'group', OWNER)).toBe(true);
expect(mayResolve(e, 'group', GLOBAL_ADMIN)).toBe(true);
expect(mayResolve(e, 'group', SCOPED_ADMIN)).toBe(false);
expect(mayResolve(e, 'group', RANDO)).toBe(false);
});
it('admins-of-scope(null) with a delivered approver keeps the delivered-to shortcut (channel semantics)', () => {
const e = { kind: 'admins-of-scope', agentGroupId: null, deliveredTo: DELIVEREE } as const;
expect(mayResolve(e, 'group', DELIVEREE)).toBe(true);
expect(mayResolve(e, 'group', SCOPED_ADMIN)).toBe(false);
});
it('D1 overlay: global scope rejects everyone below owner/global admin', () => {
const e = { kind: 'admins-of-scope', agentGroupId: 'ag-1', deliveredTo: DELIVEREE } as const;
expect(mayResolve(e, 'global', SCOPED_ADMIN)).toBe(false); // the D1 exploit, closed
expect(mayResolve(e, 'global', DELIVEREE)).toBe(false);
expect(mayResolve(e, 'global', OTHER_ADMIN)).toBe(false);
expect(mayResolve(e, 'global', OWNER)).toBe(true);
expect(mayResolve(e, 'global', GLOBAL_ADMIN)).toBe(true);
});
it('approverRuleOf maps row columns onto the rule', () => {
const base = { agent_group_id: 'ag-1' };
expect(approverRuleOf({ ...base, approver_rule: 'exclusive', approver_user_id: DELIVEREE })).toEqual({
kind: 'exclusive',
approverUserId: DELIVEREE,
});
expect(approverRuleOf({ ...base, approver_rule: 'admins-of-scope', approver_user_id: DELIVEREE })).toEqual({
kind: 'admins-of-scope',
agentGroupId: 'ag-1',
deliveredTo: DELIVEREE,
});
expect(approverRuleOf({ ...base, approver_rule: 'admins-of-scope', approver_user_id: null })).toEqual({
kind: 'admins-of-scope',
agentGroupId: 'ag-1',
deliveredTo: null,
});
// Malformed exclusive (no named user) falls back to the admin chain
// instead of bricking the hold.
expect(approverRuleOf({ ...base, approver_rule: 'exclusive', approver_user_id: null })).toEqual({
kind: 'admins-of-scope',
agentGroupId: 'ag-1',
deliveredTo: null,
});
});
});
describe('D1 regression — global-blast hold through the real response handler', () => {
beforeEach(() => {
createSession({
id: 'sess-1',
agent_group_id: 'ag-1',
messaging_group_id: null,
thread_id: null,
agent_provider: null,
status: 'active',
container_status: 'stopped',
last_active: now(),
created_at: now(),
});
initSessionFolder('ag-1', 'sess-1');
});
it("a scoped admin's click on a roles-grant-style hold is ignored; the owner's click resolves it", async () => {
const applied: string[] = [];
registerApprovalHandler('test_global_blast', async ({ userId }) => {
applied.push(userId);
});
createPendingApproval({
approval_id: 'appr-global-1',
session_id: 'sess-1',
request_id: 'appr-global-1',
action: 'test_global_blast',
payload: JSON.stringify({}),
created_at: now(),
agent_group_id: 'ag-1',
title: 'CLI: roles-grant',
options_json: JSON.stringify([]),
approver_scope: 'global',
});
// Scoped admin of the requesting group clicks approve — pre-D1 this
// resolved a global privilege grant; now it is ignored and the hold stays.
const claimedByScoped = await handleApprovalsResponse({
questionId: 'appr-global-1',
value: 'approve',
userId: 'scoped-admin',
channelType: 'slack',
platformId: 'dm-scoped',
threadId: null,
});
expect(claimedByScoped).toBe(true);
expect(applied).toEqual([]);
expect(getPendingApproval('appr-global-1')).toBeDefined();
// The owner's click resolves it.
await handleApprovalsResponse({
questionId: 'appr-global-1',
value: 'approve',
userId: 'owner',
channelType: 'slack',
platformId: 'dm-owner',
threadId: null,
});
expect(applied).toEqual([OWNER]);
expect(getPendingApproval('appr-global-1')).toBeUndefined();
});
});
-68
View File
@@ -1,68 +0,0 @@
/**
* Approver rules the one click-authorization rule for every hold.
*
* The hold-record contract (guarded-actions phase 1) is carried on the
* existing tables: a hold has an id, an action, a payload, an approver
* rule (who may resolve it), an approver scope (the action's blast radius),
* a restart policy, and an optional expiry. On `pending_approvals` these map
* to `approval_id` / `action` / `payload` / (`approver_rule` +
* `approver_user_id` + `agent_group_id`) / `approver_scope` / `expires_at`;
* the restart policy is derived from the action (`onecli_credential` rows are
* swept-and-denied on boot, everything else is durable and keeps waiting).
* `pending_channel_approvals` maps through a synthesized view
* (channel-approval.ts) the channel flow keeps its own table.
*
* Two approver-rule kinds:
* - `exclusive` only the named user may resolve (an a2a message policy's
* approver). Nobody else, including owners.
* - `admins-of-scope` the admin chain of the anchoring agent group
* (scoped admin / global admin / owner), or owners + global admins when
* the anchor is null. When the hold records the user the card was
* delivered to, that user may also resolve the sender/channel
* "named-or-admin" semantic, preserved verbatim from the pre-fold tables.
*
* The approver-scope overlay is the D1 fix: a hold whose action has global
* blast radius (e.g. `roles grant`) can only be resolved by an owner or
* global admin a scoped admin's click is rejected regardless of the
* approver rule.
*
* `mayResolve` replaces the three divergent click-auth copies (approvals
* response handler, sender handler, channel handler) with one function.
*/
import type { ApproverRule, ApproverScope, PendingApproval } from '../../types.js';
import { hasAdminPrivilege, isGlobalAdmin, isOwner } from '../permissions/db/user-roles.js';
export type { ApproverRule, ApproverScope } from '../../types.js';
/** May `clickerUserId` (namespaced `<channel>:<handle>`) resolve a hold with this approver rule + scope? */
export function mayResolve(e: ApproverRule, scope: ApproverScope, clickerUserId: string | null): boolean {
if (!clickerUserId) return false;
const globalScopeOk = scope !== 'global' || isOwner(clickerUserId) || isGlobalAdmin(clickerUserId);
if (e.kind === 'exclusive') {
return clickerUserId === e.approverUserId && globalScopeOk;
}
const eligible =
(e.deliveredTo !== null && clickerUserId === e.deliveredTo) ||
(e.agentGroupId
? hasAdminPrivilege(clickerUserId, e.agentGroupId)
: isOwner(clickerUserId) || isGlobalAdmin(clickerUserId));
return eligible && globalScopeOk;
}
/** The approver rule a `pending_approvals` row encodes. */
export function approverRuleOf(
approval: Pick<PendingApproval, 'approver_rule' | 'approver_user_id' | 'agent_group_id'>,
): ApproverRule {
if (approval.approver_rule === 'exclusive' && approval.approver_user_id) {
return { kind: 'exclusive', approverUserId: approval.approver_user_id };
}
return {
kind: 'admins-of-scope',
agentGroupId: approval.agent_group_id,
deliveredTo: approval.approver_rule === 'exclusive' ? null : approval.approver_user_id,
};
}
+14 -19
View File
@@ -25,31 +25,26 @@ import { notifyApprovalResolved } from './primitive.js';
* attribution the why, not the who (the rejecting admin may belong to a
* different owner than the requesting agent). Callers are responsible for
* clamping the reason length before passing it in.
*
* `session` is null for sessionless holds (e.g. sender admission) there is
* no agent to notify or wake, so only the row delete + resolved callbacks run.
*/
export async function finalizeReject(
approval: PendingApproval,
session: Session | null,
session: Session,
userId: string,
reason?: string,
): Promise<void> {
if (session) {
const text = reason
? `Your ${approval.action} request was rejected by admin: "${reason}"`
: `Your ${approval.action} request was rejected by admin.`;
const text = reason
? `Your ${approval.action} request was rejected by admin: "${reason}"`
: `Your ${approval.action} request was rejected by admin.`;
writeSessionMessage(session.agent_group_id, session.id, {
id: `appr-note-${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, sender: 'system', senderId: 'system' }),
});
}
writeSessionMessage(session.agent_group_id, session.id, {
id: `appr-note-${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, sender: 'system', senderId: 'system' }),
});
log.info('Approval rejected', {
approvalId: approval.approval_id,
@@ -60,5 +55,5 @@ export async function finalizeReject(
deletePendingApproval(approval.approval_id);
await notifyApprovalResolved({ approval, session, outcome: 'reject', userId });
if (session) await wakeContainer(session);
await wakeContainer(session);
}
+3 -25
View File
@@ -16,21 +16,15 @@
*
* Startup sweep edits any leftover cards from a previous process to
* "Expired (host restarted)" and drops the rows.
*
* Hold creation and every resolution (click / expiry / sweep) announce
* through the shared approval observers (notifyApprovalRequested /
* notifyApprovalResolved) observers see the full OneCLI lifecycle without
* touching this file.
*/
import { OneCLI, type ApprovalRequest, type ManualApprovalHandle } from '@onecli-sh/sdk';
import { notifyApprovalRequested, notifyApprovalResolved, pickApprovalDelivery, pickApprover } from './primitive.js';
import { pickApprovalDelivery, pickApprover } from './primitive.js';
import { ONECLI_API_KEY, ONECLI_URL } from '../../config.js';
import { getAgentGroup } from '../../db/agent-groups.js';
import {
createPendingApproval,
deletePendingApproval,
getPendingApproval,
getPendingApprovalsByAction,
updatePendingApprovalStatus,
} from '../../db/sessions.js';
@@ -70,29 +64,20 @@ function shortApprovalId(): string {
return `oa-${Math.random().toString(36).slice(2, 10)}`;
}
/** Called from the approvals response handler when a card button is clicked. `resolvedBy` = namespaced clicker id. */
export function resolveOneCLIApproval(approvalId: string, selectedOption: string, resolvedBy = ''): boolean {
/** Called from the approvals response handler when a card button is clicked. */
export function resolveOneCLIApproval(approvalId: string, selectedOption: string): boolean {
const state = pending.get(approvalId);
if (!state) return false;
pending.delete(approvalId);
clearTimeout(state.timer);
const decision: Decision = selectedOption === 'approve' ? 'approve' : 'deny';
const row = getPendingApproval(approvalId);
updatePendingApprovalStatus(approvalId, decision === 'approve' ? 'approved' : 'rejected');
// Card is auto-edited to "✅ <option>" by chat-sdk-bridge's onAction handler,
// so we don't need to deliver an edit here.
deletePendingApproval(approvalId);
state.resolve(decision);
if (row) {
void notifyApprovalResolved({
approval: row,
session: null,
outcome: decision === 'approve' ? 'approve' : 'reject',
userId: resolvedBy,
});
}
log.info('OneCLI approval resolved', { approvalId, decision });
return true;
}
@@ -210,11 +195,6 @@ async function handleRequest(request: ApprovalRequest): Promise<Decision> {
options_json: JSON.stringify(onecliOptions),
});
const created = getPendingApproval(approvalId);
if (created) {
await notifyApprovalRequested({ approval: created, session: null, deliveredTo: target.userId });
}
// Expiry timer fires just before the gateway's own TTL so our decision lands
// in time to be recorded, even though the HTTP side will already be closing.
const expiresAtMs = new Date(request.expiresAt).getTime();
@@ -242,7 +222,6 @@ async function expireApproval(approvalId: string, reason: string): Promise<void>
updatePendingApprovalStatus(approvalId, 'expired');
await editCardExpired(row, reason);
deletePendingApproval(approvalId);
await notifyApprovalResolved({ approval: row, session: null, outcome: 'expire', userId: '' });
log.info('OneCLI approval expired', { approvalId, reason });
}
@@ -272,7 +251,6 @@ async function sweepStaleApprovals(): Promise<void> {
for (const row of rows) {
await editCardExpired(row, 'host restarted');
deletePendingApproval(row.approval_id);
await notifyApprovalResolved({ approval: row, session: null, outcome: 'sweep', userId: '' });
}
}
+23 -132
View File
@@ -23,12 +23,7 @@
*/
import { normalizeOptions, type RawOption } from '../../channels/ask-question.js';
import { getMessagingGroup } from '../../db/messaging-groups.js';
import {
createPendingApproval,
getPendingApproval,
getPendingApprovalByDedupKey,
getSession,
} from '../../db/sessions.js';
import { createPendingApproval, getSession } from '../../db/sessions.js';
import { getDeliveryAdapter } from '../../delivery.js';
import { wakeContainer } from '../../container-runner.js';
import { log } from '../../log.js';
@@ -62,12 +57,11 @@ const APPROVAL_OPTIONS: RawOption[] = [
// their `requestApproval()` calls.
export interface ApprovalHandlerContext {
/** Requesting agent's session. Null for sessionless holds (e.g. sender admission). */
session: Session | null;
session: Session;
payload: Record<string, unknown>;
/** User ID of the admin who approved. Empty string if unknown. */
userId: string;
/** Send a system chat message to the requesting agent's session. No-op when sessionless. */
/** Send a system chat message to the requesting agent's session. */
notify: (text: string) => void;
}
@@ -94,20 +88,14 @@ export function getApprovalHandler(action: string): ApprovalHandler | undefined
// out. Callback errors are logged and isolated; they never block resolution.
//
// Only authorized clicks resolve an approval (the response handler's
// mayResolve gate runs first), so callbacks never fire for unauthorized
// responses. Non-click resolutions (OneCLI expiry timers, the boot sweep)
// announce here too, with outcome 'expire' / 'sweep'.
// isAuthorizedApprovalClick gate runs first), so callbacks never fire for
// unauthorized responses.
export interface ApprovalResolvedEvent {
/**
* The resolved hold. For holds that live outside pending_approvals
* (channel registration) this is a synthesized view of the same shape.
*/
approval: PendingApproval;
/** Requesting agent's session; null for sessionless holds (sender admission, OneCLI, channel registration). */
session: Session | null;
outcome: 'approve' | 'reject' | 'expire' | 'sweep';
/** Namespaced user ID (`<channel>:<handle>`) of the resolving admin. Empty string if unknown (expiry/sweep). */
session: Session;
outcome: 'approve' | 'reject';
/** Namespaced user ID (`<channel>:<handle>`) of the resolving admin. Empty string if unknown. */
userId: string;
}
@@ -136,51 +124,6 @@ export async function notifyApprovalResolved(event: ApprovalResolvedEvent): Prom
}
}
// ── Approval-requested callbacks ──
// The creation-side sibling of the resolved observer: fires once whenever a
// hold record comes into existence, whichever stack created it —
// requestApproval (cli_command, create_agent, self-mod, a2a, sender
// admission), the OneCLI credential bridge (its own rows, ids and card), and
// channel registration (as a synthesized hold view). Together with
// notifyApprovalResolved this gives observers the full hold lifecycle with
// zero touch points inside the flows.
export interface ApprovalRequestedEvent {
/**
* The created hold. For holds that live outside pending_approvals
* (channel registration) this is a synthesized view of the same shape.
*/
approval: PendingApproval;
/** Requesting agent's session; null for sessionless holds (sender admission, OneCLI, channel registration). */
session: Session | null;
/** Namespaced user ID (`<channel>:<handle>`) of the approver the card was delivered to. */
deliveredTo: string;
}
export type ApprovalRequestedHandler = (event: ApprovalRequestedEvent) => Promise<void> | void;
const approvalRequestedHandlers: ApprovalRequestedHandler[] = [];
export function registerApprovalRequestedHandler(handler: ApprovalRequestedHandler): void {
approvalRequestedHandlers.push(handler);
}
/** Fire every registered approval-requested callback. Called wherever a hold record is created. */
export async function notifyApprovalRequested(event: ApprovalRequestedEvent): Promise<void> {
for (const handler of approvalRequestedHandlers) {
try {
await handler(event);
// eslint-disable-next-line no-catch-all/no-catch-all -- isolation is the contract: one bad callback must not block the hold or other callbacks
} catch (err) {
log.error('Approval-requested handler threw', {
approvalId: event.approval.approval_id,
action: event.approval.action,
err,
});
}
}
}
// ── Approver picking ──
/**
@@ -257,14 +200,7 @@ export function notifyAgent(session: Session, text: string): void {
}
export interface RequestApprovalOptions {
/**
* Requesting agent's session. Omit for sessionless holds (e.g. sender
* admission) failure notices are then logged instead of chat-relayed,
* and the hold anchors to `agentGroupId`.
*/
session?: Session;
/** Approver-rule anchor when there is no session. Ignored when `session` is set (its agent group wins). */
agentGroupId?: string;
session: Session;
agentName: string;
/** Free-form action identifier. Must match the key the consumer registered via registerApprovalHandler. */
action: string;
@@ -274,32 +210,8 @@ export interface RequestApprovalOptions {
title: string;
/** Card body shown to the admin. */
question: string;
/**
* Deliver the card to this specific user AND make the hold exclusively
* theirs to resolve (approver rule 'exclusive' an a2a policy's approver).
*/
/** Deliver the card to this specific user instead of all of the session group's admins. */
approverUserId?: string;
/**
* The action's blast radius. 'global' holds (privilege grants, cross-group
* writes) can only be resolved by an owner or global admin. Default 'group'.
*/
approverScope?: 'group' | 'global';
/** Card buttons. Default: Approve / Reject / Reject with reason…. */
options?: RawOption[];
/**
* In-flight dedup: while a pending row carries this key, a repeat request
* with the same key is dropped without a second card.
*/
dedupKey?: string;
/**
* Record the user the card is delivered to on the hold, letting them
* resolve it alongside the scope's admins even if their role changes
* mid-flight (the sender/channel "named-or-admin" semantic). Off by
* default module holds authorize purely by the admin chain.
*/
recordDeliveredApprover?: boolean;
/** Channel preference for the approver-DM walk when there is no session to derive it from. */
originChannelType?: string;
}
/**
@@ -309,59 +221,38 @@ export interface RequestApprovalOptions {
* approval handler for this action via the response dispatcher.
*/
export async function requestApproval(opts: RequestApprovalOptions): Promise<void> {
const { session, action, payload, title, question, agentName, approverUserId, dedupKey } = opts;
const { session, action, payload, title, question, agentName, approverUserId } = opts;
const agentGroupId = session?.agent_group_id ?? opts.agentGroupId ?? null;
const fail = (text: string): void => {
if (session) notifyAgent(session, `${action} failed: ${text}`);
else log.warn('Approval request failed', { action, agentGroupId, reason: text });
};
if (dedupKey && getPendingApprovalByDedupKey(dedupKey)) {
log.debug('Approval request already in flight — dropping duplicate', { action, dedupKey });
return;
}
const approvers = approverUserId ? [approverUserId] : pickApprover(agentGroupId);
const approvers = approverUserId ? [approverUserId] : pickApprover(session.agent_group_id);
if (approvers.length === 0) {
fail('no owner or admin configured to approve.');
notifyAgent(session, `${action} failed: no owner or admin configured to approve.`);
return;
}
const originChannelType =
opts.originChannelType ??
(session?.messaging_group_id ? (getMessagingGroup(session.messaging_group_id)?.channel_type ?? '') : '');
const originChannelType = session.messaging_group_id
? (getMessagingGroup(session.messaging_group_id)?.channel_type ?? '')
: '';
const target = await pickApprovalDelivery(approvers, originChannelType);
if (!target) {
fail('no DM channel found for any eligible approver.');
notifyAgent(session, `${action} failed: no DM channel found for any eligible approver.`);
return;
}
const approvalId = `appr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const cardOptions = opts.options ?? APPROVAL_OPTIONS;
const normalizedOptions = normalizeOptions(APPROVAL_OPTIONS);
createPendingApproval({
approval_id: approvalId,
session_id: session?.id ?? null,
session_id: session.id,
request_id: approvalId,
action,
payload: JSON.stringify(payload),
created_at: new Date().toISOString(),
agent_group_id: agentGroupId,
title,
options_json: JSON.stringify(normalizeOptions(cardOptions)),
approver_user_id: approverUserId ?? (opts.recordDeliveredApprover ? target.userId : null),
approver_rule: approverUserId ? 'exclusive' : 'admins-of-scope',
approver_scope: opts.approverScope ?? 'group',
dedup_key: dedupKey ?? null,
options_json: JSON.stringify(normalizedOptions),
approver_user_id: approverUserId ?? null,
});
const created = getPendingApproval(approvalId);
if (created) {
await notifyApprovalRequested({ approval: created, session: session ?? null, deliveredTo: target.userId });
}
const adapter = getDeliveryAdapter();
if (adapter) {
try {
@@ -375,12 +266,12 @@ export async function requestApproval(opts: RequestApprovalOptions): Promise<voi
questionId: approvalId,
title,
question,
options: cardOptions,
options: APPROVAL_OPTIONS,
}),
);
} catch (err) {
log.error('Failed to deliver approval card', { action, approvalId, err });
fail(`could not deliver approval request to ${target.userId}.`);
notifyAgent(session, `${action} failed: could not deliver approval request to ${target.userId}.`);
return;
}
}
+44 -32
View File
@@ -12,9 +12,6 @@
* 2. OneCLI credential approvals (`action = 'onecli_credential'`). Resolved
* via an in-memory Promise see onecli-approvals.ts.
*
* Click authorization is `mayResolve` over the hold's approver rule +
* approver scope (approver-rule.ts) the one shared rule for every hold.
*
* The response handler is registered via core's `registerResponseHandler`;
* core iterates handlers and the first one to return `true` claims the response.
*/
@@ -23,8 +20,8 @@ import { deletePendingApproval, getPendingApproval, getSession } from '../../db/
import type { ResponsePayload } from '../../response-registry.js';
import { log } from '../../log.js';
import { writeSessionMessage } from '../../session-manager.js';
import type { PendingApproval, Session } from '../../types.js';
import { approverRuleOf, mayResolve } from './approver-rule.js';
import type { PendingApproval } from '../../types.js';
import { hasAdminPrivilege, isGlobalAdmin, isOwner } from '../permissions/db/user-roles.js';
import { finalizeReject } from './finalize.js';
import { ONECLI_ACTION, resolveOneCLIApproval } from './onecli-approvals.js';
import { getApprovalHandler, notifyApprovalResolved, REJECT_WITH_REASON_VALUE } from './primitive.js';
@@ -34,8 +31,7 @@ export async function handleApprovalsResponse(payload: ResponsePayload): Promise
const approval = getPendingApproval(payload.questionId);
if (!approval) return false;
const clickerId = namespacedUserId(payload);
if (!mayResolve(approverRuleOf(approval), approval.approver_scope, clickerId)) {
if (!isAuthorizedApprovalClick(approval, payload)) {
log.warn('Ignoring unauthorized approval response', {
approvalId: approval.approval_id,
action: approval.action,
@@ -46,7 +42,7 @@ export async function handleApprovalsResponse(payload: ResponsePayload): Promise
}
if (approval.action === ONECLI_ACTION) {
if (resolveOneCLIApproval(payload.questionId, payload.value, clickerId ?? '')) {
if (resolveOneCLIApproval(payload.questionId, payload.value)) {
return true;
}
// Row exists but the in-memory resolver is gone (timer fired or the process
@@ -55,7 +51,7 @@ export async function handleApprovalsResponse(payload: ResponsePayload): Promise
return true;
}
await handleRegisteredApproval(approval, payload.value, clickerId ?? '');
await handleRegisteredApproval(approval, payload.value, namespacedUserId(payload) ?? '');
return true;
}
@@ -64,11 +60,12 @@ async function handleRegisteredApproval(
selectedOption: string,
userId: string,
): Promise<void> {
// Sessionless holds (sender admission) carry session_id null and resolve
// without an agent to notify; a session-BOUND hold whose session vanished
// is stale — drop it.
const session: Session | null = approval.session_id ? (getSession(approval.session_id) ?? null) : null;
if (approval.session_id && !session) {
if (!approval.session_id) {
deletePendingApproval(approval.approval_id);
return;
}
const session = getSession(approval.session_id);
if (!session) {
deletePendingApproval(approval.approval_id);
return;
}
@@ -76,10 +73,8 @@ async function handleRegisteredApproval(
// "Reject with reason…" — hold the row and capture the admin's next DM
// instead of finalizing now. The agent is notified exactly once: after the
// reason arrives, or after the sweep's timeout if the admin ghosts.
// Sessionless holds have nobody to relay a reason to — plain reject.
if (selectedOption === REJECT_WITH_REASON_VALUE) {
if (session) await armReasonCapture(approval, session, userId);
else await finalizeReject(approval, null, userId);
await armReasonCapture(approval, session, userId);
return;
}
@@ -90,19 +85,17 @@ async function handleRegisteredApproval(
}
// Approved — dispatch to the module that registered for this action.
const notify = session
? (text: string): void => {
writeSessionMessage(session.agent_group_id, session.id, {
id: `appr-note-${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, sender: 'system', senderId: 'system' }),
});
}
: (): void => {};
const notify = (text: string): void => {
writeSessionMessage(session.agent_group_id, session.id, {
id: `appr-note-${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, sender: 'system', senderId: 'system' }),
});
};
const handler = getApprovalHandler(approval.action);
if (!handler) {
@@ -113,7 +106,7 @@ async function handleRegisteredApproval(
notify(`Your ${approval.action} was approved, but no handler is installed to apply it.`);
deletePendingApproval(approval.approval_id);
await notifyApprovalResolved({ approval, session, outcome: 'approve', userId });
if (session) await wakeContainer(session);
await wakeContainer(session);
return;
}
@@ -130,10 +123,29 @@ async function handleRegisteredApproval(
deletePendingApproval(approval.approval_id);
await notifyApprovalResolved({ approval, session, outcome: 'approve', userId });
if (session) await wakeContainer(session);
await wakeContainer(session);
}
function namespacedUserId(payload: ResponsePayload): string | null {
if (!payload.userId) return null;
return payload.userId.includes(':') ? payload.userId : `${payload.channelType}:${payload.userId}`;
}
function isAuthorizedApprovalClick(approval: PendingApproval, payload: ResponsePayload): boolean {
const userId = namespacedUserId(payload);
if (!userId) return false;
// An approval may name a specific approver; only that exact user may resolve it.
if (approval.approver_user_id) {
return userId === approval.approver_user_id;
}
const agentGroupId =
approval.agent_group_id ?? (approval.session_id ? getSession(approval.session_id)?.agent_group_id : null);
if (!agentGroupId) {
return isOwner(userId) || isGlobalAdmin(userId);
}
return hasAdminPrivilege(userId, agentGroupId);
}
-1
View File
@@ -18,7 +18,6 @@
// registers its handlers at import time.
import './approvals/index.js';
import './interactive/index.js';
import './scheduling/index.js';
import './permissions/index.js';
import './agent-to-agent/index.js';
import './self-mod/index.js';
@@ -176,23 +176,6 @@ describe('unknown-channel registration flow', () => {
expect(count).toBe(1);
});
it('announces the hold through the shared approval-requested observer', async () => {
const { registerApprovalRequestedHandler } = await import('../approvals/primitive.js');
const { routeInbound } = await import('../../router.js');
const events: Array<{ approval: { approval_id: string; action: string }; deliveredTo: string }> = [];
registerApprovalRequestedHandler((event) => {
if (event.approval.action === 'channel_registration') events.push(event);
});
await routeInbound(groupMention('chat-observed'));
await new Promise((r) => setTimeout(r, 10));
expect(events).toHaveLength(1);
expect(events[0].deliveredTo).toBe('telegram:owner');
expect(events[0].approval.approval_id).toMatch(/^mg-/); // the hold view is keyed by the messaging group
});
it('dedups a second mention while the card is pending', async () => {
const { routeInbound } = await import('../../router.js');
await routeInbound(groupMention('chat-busy'));
@@ -375,7 +358,7 @@ describe('unknown-channel registration flow', () => {
expect(stillPending).toBe(1);
});
it('does not let a scoped admin drive channel registration at all (D4: owner/global-admin approver rule)', async () => {
it('does not let a scoped admin connect an unknown channel to another agent group', async () => {
const { routeInbound } = await import('../../router.js');
const { getResponseHandlers } = await import('../../response-registry.js');
const { getDb } = await import('../../db/connection.js');
@@ -412,28 +395,39 @@ describe('unknown-channel registration flow', () => {
messaging_group_id: string;
};
expect(pending).toBeDefined();
// Registration creates groups/wirings — the card goes to the global chain
// (the owner's DM), never to a scoped admin, even though one exists.
expect(deliverMock).toHaveBeenCalledTimes(1);
expect(deliverMock.mock.calls[0][1]).toBe('dm-owner');
expect(deliverMock.mock.calls[0][1]).toBe('dm-scoped-admin');
// A scoped admin's clicks are ignored outright — no follow-up card, no
// wiring, and the pending row stays for a real approver.
for (const value of ['choose_existing', 'connect:ag-2', `connect:ag-1`]) {
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending.messaging_group_id,
value,
userId: 'scoped-admin',
channelType: 'telegram',
platformId: 'dm-scoped-admin',
threadId: null,
});
if (claimed) break;
}
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending.messaging_group_id,
value: 'choose_existing',
userId: 'scoped-admin',
channelType: 'telegram',
platformId: 'dm-scoped-admin',
threadId: null,
});
if (claimed) break;
}
const followupPayload = JSON.parse(deliverMock.mock.calls[1][4] as string) as {
options: Array<{ label: string; value: string }>;
};
expect(followupPayload.options.map((option) => option.value)).toContain('connect:ag-1');
expect(followupPayload.options.map((option) => option.value)).not.toContain('connect:ag-2');
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending.messaging_group_id,
value: 'connect:ag-2',
userId: 'scoped-admin',
channelType: 'telegram',
platformId: 'dm-scoped-admin',
threadId: null,
});
if (claimed) break;
}
expect(deliverMock).toHaveBeenCalledTimes(1); // no follow-up card was sent
const mgaCount = (
getDb()
.prepare('SELECT COUNT(*) AS c FROM messaging_group_agents WHERE messaging_group_id = ?')
+13 -51
View File
@@ -52,13 +52,9 @@ import { getDeliveryAdapter } from '../../delivery.js';
import { initGroupFilesystem } from '../../group-init.js';
import { log } from '../../log.js';
import type { InboundEvent } from '../../channels/adapter.js';
import type { AgentGroup, PendingApproval } from '../../types.js';
import { notifyApprovalRequested, pickApprovalDelivery, pickApprover } from '../approvals/primitive.js';
import {
createPendingChannelApproval,
hasInFlightChannelApproval,
type PendingChannelApproval,
} from './db/pending-channel-approvals.js';
import type { AgentGroup } from '../../types.js';
import { pickApprovalDelivery, pickApprover } from '../approvals/primitive.js';
import { createPendingChannelApproval, hasInFlightChannelApproval } from './db/pending-channel-approvals.js';
import { hasAdminPrivilege } from './db/user-roles.js';
// ── Value constants (response handler in index.ts parses these) ──
@@ -156,14 +152,15 @@ export async function requestChannelApproval(input: RequestChannelApprovalInput)
});
return;
}
// Use first agent group for approver resolution — owners and global admins
// are returned regardless of which group we pass.
const referenceGroup = agentGroups[0];
// Registration creates groups and wirings — global blast radius, so the
// approver comes from the global chain (global admins → owners), never from
// whichever agent group happens to sort first.
const approvers = pickApprover(null);
const approvers = pickApprover(referenceGroup.id);
if (approvers.length === 0) {
log.warn('Channel registration skipped — no owner or global admin configured', {
log.warn('Channel registration skipped — no owner or admin configured', {
messagingGroupId,
targetAgentGroupId: referenceGroup.id,
});
return;
}
@@ -191,6 +188,7 @@ export async function requestChannelApproval(input: RequestChannelApprovalInput)
if (!delivery) {
log.warn('Channel registration skipped — no DM channel for any approver', {
messagingGroupId,
targetAgentGroupId: referenceGroup.id,
});
return;
}
@@ -210,19 +208,15 @@ export async function requestChannelApproval(input: RequestChannelApprovalInput)
const question = buildQuestionText(isGroup, senderName, channelName, originChannelType);
const options = normalizeOptions(buildApprovalOptions(agentGroups, delivery.userId));
// agent_group_id is NOT NULL bookkeeping (the schema predates the global
// approver chain); it no longer drives approver resolution or click-auth.
const row: PendingChannelApproval = {
createPendingChannelApproval({
messaging_group_id: messagingGroupId,
agent_group_id: agentGroups[0].id,
agent_group_id: referenceGroup.id,
original_message: JSON.stringify(event),
approver_user_id: delivery.userId,
created_at: new Date().toISOString(),
title,
options_json: JSON.stringify(options),
};
createPendingChannelApproval(row);
await notifyApprovalRequested({ approval: channelHoldView(row), session: null, deliveredTo: delivery.userId });
});
const adapter = getDeliveryAdapter();
if (!adapter) {
@@ -277,38 +271,6 @@ export function buildAgentSelectionOptions(
return normalizeOptions(options);
}
/**
* The channel-registration hold as a hold-record view (the shape
* pending_approvals rows have), so its terminal resolutions can announce
* through the shared approval-resolved observer. The flow itself keeps its
* own table and multi-step conversation.
*/
export function channelHoldView(
row: PendingChannelApproval,
payloadExtra: Record<string, unknown> = {},
): PendingApproval {
return {
approval_id: row.messaging_group_id,
session_id: null,
request_id: row.messaging_group_id,
action: 'channel_registration',
payload: JSON.stringify({ messagingGroupId: row.messaging_group_id, ...payloadExtra }),
created_at: row.created_at,
agent_group_id: null,
channel_type: null,
platform_id: null,
platform_message_id: null,
expires_at: null,
status: 'pending',
title: row.title,
options_json: row.options_json,
approver_user_id: row.approver_user_id,
approver_rule: 'admins-of-scope',
approver_scope: 'group',
dedup_key: null,
};
}
/**
* Create a new agent group and initialize its filesystem. Handles
* folder-name collisions with numeric suffixes.
@@ -0,0 +1,60 @@
/**
* CRUD for pending_sender_approvals the in-flight state for the
* request_approval unknown-sender flow. Rows are created when an unknown
* sender writes into a wired messaging group with that policy, and are
* deleted on admin approve (after adding the user as a member) or deny.
*
* UNIQUE(messaging_group_id, sender_identity) enforces in-flight dedup:
* a retry / second message from the same unknown sender while a card is
* still pending is silently dropped instead of spamming the admin.
*/
import { getDb } from '../../../db/connection.js';
export interface PendingSenderApproval {
id: string;
messaging_group_id: string;
agent_group_id: string;
sender_identity: string;
sender_name: string | null;
original_message: string;
approver_user_id: string;
created_at: string;
/** Card title shown at creation and re-used by getAskQuestionRender on click. */
title: string;
/** Normalized options (JSON-encoded NormalizedOption[]) — same shape persisted on pending_approvals. */
options_json: string;
}
export function createPendingSenderApproval(row: PendingSenderApproval): void {
getDb()
.prepare(
`INSERT INTO pending_sender_approvals (
id, messaging_group_id, agent_group_id, sender_identity,
sender_name, original_message, approver_user_id, created_at,
title, options_json
)
VALUES (
@id, @messaging_group_id, @agent_group_id, @sender_identity,
@sender_name, @original_message, @approver_user_id, @created_at,
@title, @options_json
)`,
)
.run(row);
}
export function getPendingSenderApproval(id: string): PendingSenderApproval | undefined {
return getDb().prepare('SELECT * FROM pending_sender_approvals WHERE id = ?').get(id) as
| PendingSenderApproval
| undefined;
}
export function hasInFlightSenderApproval(messagingGroupId: string, senderIdentity: string): boolean {
const row = getDb()
.prepare('SELECT 1 AS x FROM pending_sender_approvals WHERE messaging_group_id = ? AND sender_identity = ?')
.get(messagingGroupId, senderIdentity) as { x: number } | undefined;
return row !== undefined;
}
export function deletePendingSenderApproval(id: string): void {
getDb().prepare('DELETE FROM pending_sender_approvals WHERE id = ?').run(id);
}
+77 -58
View File
@@ -32,12 +32,9 @@ import { registerResponseHandler, type ResponsePayload } from '../../response-re
import { getDeliveryAdapter } from '../../delivery.js';
import { log } from '../../log.js';
import type { MessagingGroup, MessagingGroupAgent } from '../../types.js';
import { mayResolve } from '../approvals/approver-rule.js';
import { notifyApprovalResolved, registerApprovalHandler } from '../approvals/primitive.js';
import { canAccessAgentGroup } from './access.js';
import {
buildAgentSelectionOptions,
channelHoldView,
CHOOSE_EXISTING_VALUE,
CONNECT_PREFIX,
createNewAgentGroup,
@@ -51,9 +48,10 @@ import {
getPendingChannelApproval,
updatePendingChannelApprovalCard,
} from './db/pending-channel-approvals.js';
import { deletePendingSenderApproval, getPendingSenderApproval } from './db/pending-sender-approvals.js';
import { hasAdminPrivilege } from './db/user-roles.js';
import { getUser, upsertUser } from './db/users.js';
import { requestSenderApproval, SENDER_ADMIT_ACTION } from './sender-approval.js';
import { requestSenderApproval } from './sender-approval.js';
import { ensureUserDm } from './user-dm.js';
// ── Free-text name input state ──
@@ -211,37 +209,83 @@ setSenderScopeGate(
);
/**
* Approve continuation for the unknown-sender hold (a sessionless
* pending_approvals row created by sender-approval.ts): add the sender to
* agent_group_members and re-invoke routeInbound with the stored event the
* second routing attempt clears the gate because the user is now a member.
* Response handler for the unknown-sender approval card.
*
* Click authorization and the deny path are the approvals module's shared
* response handler (mayResolve + finalizeReject). Deny just drops the hold
* no "deny list"; a future message re-triggers a fresh card.
* Claim rule: questionId matches a row in pending_sender_approvals. If no
* such row, return false so the next handler (approvals module, OneCLI,
* interactive) gets a shot.
*
* Approve: add the sender to agent_group_members + re-invoke routeInbound
* with the stored event. The second routing attempt clears the gate because
* the user is now a member.
*
* Deny: delete the row (no "deny list" a future message re-triggers a
* fresh card per ACTION-ITEMS item 5 "no denial persistence").
*/
registerApprovalHandler(SENDER_ADMIT_ACTION, async ({ payload, userId }) => {
const senderIdentity = typeof payload.senderIdentity === 'string' ? payload.senderIdentity : '';
const agentGroupId = typeof payload.agentGroupId === 'string' ? payload.agentGroupId : '';
if (!senderIdentity || !agentGroupId) {
log.warn('sender_admit approved but the hold payload was malformed', { senderIdentity, agentGroupId });
return;
async function handleSenderApprovalResponse(payload: ResponsePayload): Promise<boolean> {
const row = getPendingSenderApproval(payload.questionId);
if (!row) return false;
// payload.userId is the raw platform userId (e.g. "6037840640"); namespace it
// with the channel type so it matches users(id) format. Some platforms
// (e.g. Teams "29:xxx") already include a colon — mirror resolveOrCreateUser
// logic and only prefix when the raw id has no colon.
const clickerId = payload.userId
? payload.userId.includes(':')
? payload.userId
: `${payload.channelType}:${payload.userId}`
: null;
const isAuthorized =
clickerId !== null && (clickerId === row.approver_user_id || hasAdminPrivilege(clickerId, row.agent_group_id));
if (!isAuthorized) {
log.warn('Unknown-sender approval click rejected — unauthorized clicker', {
approvalId: row.id,
clickerId,
expectedApprover: row.approver_user_id,
});
return true; // claim the response so it's not unclaimed-logged, but do nothing
}
const approverId = clickerId;
const approved = payload.value === 'approve';
if (approved) {
addMember({
user_id: row.sender_identity,
agent_group_id: row.agent_group_id,
added_by: approverId,
added_at: new Date().toISOString(),
});
log.info('Unknown sender approved — member added', {
approvalId: row.id,
senderIdentity: row.sender_identity,
agentGroupId: row.agent_group_id,
approverId,
});
// Clear the pending row BEFORE re-routing so the gate check on the
// second attempt doesn't see the in-flight row and short-circuit.
deletePendingSenderApproval(row.id);
try {
const event = JSON.parse(row.original_message) as InboundEvent;
await routeInbound(event);
} catch (err) {
log.error('Failed to replay message after sender approval', { approvalId: row.id, err });
}
return true;
}
addMember({
user_id: senderIdentity,
agent_group_id: agentGroupId,
added_by: userId,
added_at: new Date().toISOString(),
log.info('Unknown sender denied', {
approvalId: row.id,
senderIdentity: row.sender_identity,
agentGroupId: row.agent_group_id,
approverId,
});
log.info('Unknown sender approved — member added', { senderIdentity, agentGroupId, approverId: userId });
deletePendingSenderApproval(row.id);
return true;
}
try {
await routeInbound(payload.event as InboundEvent);
} catch (err) {
log.error('Failed to replay message after sender approval', { senderIdentity, agentGroupId, err });
}
});
registerResponseHandler(handleSenderApprovalResponse);
// ── Unknown-channel registration flow ──
@@ -267,21 +311,14 @@ async function handleChannelApprovalResponse(payload: ResponsePayload): Promise<
const row = getPendingChannelApproval(payload.questionId);
if (!row) return false;
// payload.userId is the raw platform userId (e.g. "6037840640"); namespace it
// with the channel type so it matches users(id) format. Some platforms
// (e.g. Teams "29:xxx") already include a colon — mirror resolveOrCreateUser
// logic and only prefix when the raw id has no colon.
const clickerId = payload.userId
? payload.userId.includes(':')
? payload.userId
: `${payload.channelType}:${payload.userId}`
: null;
// Channel registration creates groups and wirings — the approver rule is
// owner / global admin (plus the delivered-to approver), never an admin
// scoped to whichever group happens to sort first.
if (
!mayResolve({ kind: 'admins-of-scope', agentGroupId: null, deliveredTo: row.approver_user_id }, 'group', clickerId)
) {
const isAuthorized =
clickerId !== null && (clickerId === row.approver_user_id || hasAdminPrivilege(clickerId, row.agent_group_id));
if (!isAuthorized) {
log.warn('Channel registration click rejected — unauthorized clicker', {
messagingGroupId: row.messaging_group_id,
clickerId,
@@ -289,18 +326,12 @@ async function handleChannelApprovalResponse(payload: ResponsePayload): Promise<
});
return true;
}
const approverId = clickerId as string;
const approverId = clickerId;
// ── Reject / Cancel ──
if (payload.value === REJECT_VALUE) {
setMessagingGroupDeniedAt(row.messaging_group_id, new Date().toISOString());
deletePendingChannelApproval(row.messaging_group_id);
await notifyApprovalResolved({
approval: channelHoldView(row),
session: null,
outcome: 'reject',
userId: approverId,
});
log.info('Channel registration denied', {
messagingGroupId: row.messaging_group_id,
approverId,
@@ -472,12 +503,6 @@ async function handleChannelApprovalResponse(payload: ResponsePayload): Promise<
}
deletePendingChannelApproval(row.messaging_group_id);
await notifyApprovalResolved({
approval: channelHoldView(row, { targetAgentGroupId }),
session: null,
outcome: 'approve',
userId: approverId,
});
try {
await routeInbound(event);
@@ -578,12 +603,6 @@ registerMessageInterceptor(async (event: InboundEvent): Promise<boolean> => {
}
deletePendingChannelApproval(row.messaging_group_id);
await notifyApprovalResolved({
approval: channelHoldView(row, { targetAgentGroupId: ag.id, createdAgentGroup: true }),
session: null,
outcome: 'approve',
userId,
});
try {
await routeInbound(originalEvent);
+36 -67
View File
@@ -1,8 +1,6 @@
/**
* Integration tests for the unknown-sender request_approval flow
* (ACTION-ITEMS item 5), folded onto the approvals primitive: the hold is a
* sessionless pending_approvals row (action 'sender_admit') resolved by the
* approvals module's shared response handler.
* (ACTION-ITEMS item 5).
*
* Covers:
* - request_approval policy fires `requestSenderApproval` on first unknown
@@ -11,9 +9,7 @@
* silently dropped (no second card, no second row)
* - Approve path: member added, original message replayed via routeInbound,
* container woken
* - Deny path: pending hold deleted, no member added
* - Click authorization: named-or-admin (delivered approver or any admin of
* the group's chain); strangers can't self-admit
* - Deny path: pending row deleted, no member added
*/
import fs from 'fs';
import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest';
@@ -32,14 +28,12 @@ vi.mock('../../container-runner.js', () => ({
killContainer: vi.fn(),
}));
// Mock delivery adapter — record card deliveries for assertions. The approvals
// barrel also pulls onDeliveryAdapterReady from this module at import time.
// Mock delivery adapter — record card deliveries for assertions.
const deliverMock = vi.fn().mockResolvedValue('plat-msg-id');
vi.mock('../../delivery.js', () => ({
getDeliveryAdapter: () => ({
deliver: deliverMock,
}),
onDeliveryAdapterReady: vi.fn(),
}));
// Mock ensureUserDm to return the approver's existing messaging group
@@ -75,12 +69,10 @@ beforeEach(async () => {
const db = initTestDb();
runMigrations(db);
// Side-effect imports: register hooks AFTER the mocks are in place so the
// access gate / response handler pick up the mocked delivery + user-dm
// helpers. The approvals barrel registers the shared response handler that
// resolves sender_admit holds.
// Side-effect imports: register hooks (permissions module) AFTER the
// mocks are in place so the access gate / response handler pick up the
// mocked delivery + user-dm helpers.
await import('./index.js');
await import('../approvals/index.js');
// Fixtures: agent group, messaging group with request_approval, wiring,
// owner + DM messaging group for approver delivery.
@@ -160,20 +152,6 @@ function stranger(text: string) {
};
}
async function pendingSenderHold(): Promise<{ approval_id: string } | undefined> {
const { getDb } = await import('../../db/connection.js');
return getDb().prepare("SELECT approval_id FROM pending_approvals WHERE action = 'sender_admit'").get() as
| { approval_id: string }
| undefined;
}
async function senderHoldCount(): Promise<number> {
const { getDb } = await import('../../db/connection.js');
return (
getDb().prepare("SELECT COUNT(*) AS c FROM pending_approvals WHERE action = 'sender_admit'").get() as { c: number }
).c;
}
describe('unknown-sender request_approval flow', () => {
it('delivers an approval card on first unknown message', async () => {
const { routeInbound } = await import('../../router.js');
@@ -190,26 +168,11 @@ describe('unknown-sender request_approval flow', () => {
expect(kind).toBe('chat-sdk');
const payload = JSON.parse(content as string);
expect(payload.type).toBe('ask_question');
expect(payload.questionId).toMatch(/^appr-/);
expect(payload.title).toBe('👤 New sender');
expect(payload.options.map((o: { value: string }) => o.value)).toEqual(['approve', 'reject']);
expect(payload.questionId).toMatch(/^nsa-/);
const { getDb } = await import('../../db/connection.js');
const rows = getDb().prepare("SELECT * FROM pending_approvals WHERE action = 'sender_admit'").all() as Array<{
session_id: string | null;
agent_group_id: string;
approver_rule: string;
approver_user_id: string;
dedup_key: string;
}>;
const rows = getDb().prepare('SELECT * FROM pending_sender_approvals').all();
expect(rows).toHaveLength(1);
// Hold-record contract: sessionless, anchored to the agent group,
// named-or-admin approver rule with the delivered approver recorded.
expect(rows[0].session_id).toBeNull();
expect(rows[0].agent_group_id).toBe('ag-1');
expect(rows[0].approver_rule).toBe('admins-of-scope');
expect(rows[0].approver_user_id).toBe('telegram:owner');
expect(rows[0].dedup_key).toBe('sender_admit:mg-chat:tg:stranger');
});
it('dedups a second message from the same stranger while pending', async () => {
@@ -220,7 +183,9 @@ describe('unknown-sender request_approval flow', () => {
await new Promise((r) => setTimeout(r, 10));
expect(deliverMock).toHaveBeenCalledTimes(1);
expect(await senderHoldCount()).toBe(1);
const { getDb } = await import('../../db/connection.js');
const count = (getDb().prepare('SELECT COUNT(*) AS c FROM pending_sender_approvals').get() as { c: number }).c;
expect(count).toBe(1);
});
it('approve → adds member and replays the original message', async () => {
@@ -232,16 +197,17 @@ describe('unknown-sender request_approval flow', () => {
await routeInbound(stranger('please let me in'));
await new Promise((r) => setTimeout(r, 10));
const pending = await pendingSenderHold();
const { getDb } = await import('../../db/connection.js');
const pending = getDb().prepare('SELECT id FROM pending_sender_approvals').get() as { id: string };
expect(pending).toBeDefined();
// Fire the approve click through the response-handler chain.
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending!.approval_id,
questionId: pending.id,
value: 'approve',
// Chat SDK's onAction surfaces the raw platform userId (e.g. Telegram
// chat id). The response handler namespaces it with channelType to
// chat id). The permissions handler namespaces it with channelType to
// match users(id).
userId: 'owner',
channelType: 'telegram',
@@ -252,32 +218,33 @@ describe('unknown-sender request_approval flow', () => {
}
// Member row added for the stranger against the wired agent group.
const { getDb } = await import('../../db/connection.js');
const member = getDb()
.prepare('SELECT 1 AS x FROM agent_group_members WHERE user_id = ? AND agent_group_id = ?')
.get('tg:stranger', 'ag-1');
expect(member).toBeDefined();
// Pending hold cleared.
expect(await senderHoldCount()).toBe(0);
// Pending row cleared.
const stillPending = getDb().prepare('SELECT COUNT(*) AS c FROM pending_sender_approvals').get() as { c: number };
expect(stillPending.c).toBe(0);
// Message replayed + container woken.
expect(wakeContainer).toHaveBeenCalled();
});
it('deny → deletes the pending hold without adding a member', async () => {
it('deny → deletes the pending row without adding a member', async () => {
const { routeInbound } = await import('../../router.js');
const { getResponseHandlers } = await import('../../response-registry.js');
await routeInbound(stranger('hello'));
await new Promise((r) => setTimeout(r, 10));
const pending = await pendingSenderHold();
const { getDb } = await import('../../db/connection.js');
const pending = getDb().prepare('SELECT id FROM pending_sender_approvals').get() as { id: string };
expect(pending).toBeDefined();
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending!.approval_id,
questionId: pending.id,
value: 'reject',
userId: 'owner', // raw platform id — handler namespaces with channelType
channelType: 'telegram',
@@ -287,8 +254,8 @@ describe('unknown-sender request_approval flow', () => {
if (claimed) break;
}
expect(await senderHoldCount()).toBe(0);
const { getDb } = await import('../../db/connection.js');
const count = (getDb().prepare('SELECT COUNT(*) AS c FROM pending_sender_approvals').get() as { c: number }).c;
expect(count).toBe(0);
const member = getDb()
.prepare('SELECT 1 AS x FROM agent_group_members WHERE user_id = ? AND agent_group_id = ?')
.get('tg:stranger', 'ag-1');
@@ -303,7 +270,8 @@ describe('unknown-sender request_approval flow', () => {
await routeInbound(stranger('can I play'));
await new Promise((r) => setTimeout(r, 10));
const pending = await pendingSenderHold();
const { getDb } = await import('../../db/connection.js');
const pending = getDb().prepare('SELECT id FROM pending_sender_approvals').get() as { id: string };
expect(pending).toBeDefined();
// A random user (not the stranger, not the owner, not an admin) tries to
@@ -311,7 +279,7 @@ describe('unknown-sender request_approval flow', () => {
// rejected without admitting them.
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending!.approval_id,
questionId: pending.id,
value: 'approve',
userId: 'random-bystander', // not owner, not admin
channelType: 'telegram',
@@ -322,17 +290,18 @@ describe('unknown-sender request_approval flow', () => {
}
// No member added for the stranger.
const { getDb } = await import('../../db/connection.js');
const member = getDb()
.prepare('SELECT 1 AS x FROM agent_group_members WHERE user_id = ? AND agent_group_id = ?')
.get('tg:stranger', 'ag-1');
expect(member).toBeUndefined();
// Pending hold is still there — a legitimate approver can still act on it.
expect(await senderHoldCount()).toBe(1);
// Pending row is still there — a legitimate approver can still act on it.
const stillPending = (getDb().prepare('SELECT COUNT(*) AS c FROM pending_sender_approvals').get() as { c: number })
.c;
expect(stillPending).toBe(1);
});
it('accepts a click from a global admin even if they are not the delivered approver', async () => {
it('accepts a click from a global admin even if they are not the designated approver', async () => {
// Pre-seed a separate admin user so we can click as them.
upsertUser({ id: 'telegram:admin-bob', kind: 'telegram', display_name: 'Bob', created_at: now() });
grantRole({
@@ -349,13 +318,14 @@ describe('unknown-sender request_approval flow', () => {
await routeInbound(stranger('knock knock'));
await new Promise((r) => setTimeout(r, 10));
const pending = await pendingSenderHold();
const { getDb } = await import('../../db/connection.js');
const pending = getDb().prepare('SELECT id FROM pending_sender_approvals').get() as { id: string };
expect(pending).toBeDefined();
// Admin clicks approve (not the delivered approver, which was the owner).
// Admin clicks approve (not the designated approver, which was owner).
for (const handler of getResponseHandlers()) {
const claimed = await handler({
questionId: pending!.approval_id,
questionId: pending.id,
value: 'approve',
userId: 'admin-bob',
channelType: 'telegram',
@@ -366,7 +336,6 @@ describe('unknown-sender request_approval flow', () => {
}
// Stranger admitted thanks to the admin's authority.
const { getDb } = await import('../../db/connection.js');
const member = getDb()
.prepare('SELECT 1 AS x FROM agent_group_members WHERE user_id = ? AND agent_group_id = ?')
.get('tg:stranger', 'ag-1');
+123 -35
View File
@@ -3,38 +3,44 @@
*
* When `messaging_groups.unknown_sender_policy = 'request_approval'` and a
* non-member writes into a wired chat, the access gate drops the routing
* attempt and calls `requestSenderApproval`, which holds through the
* approvals primitive (action 'sender_admit'):
* attempt and calls `requestSenderApproval` to:
*
* - approver rule: the agent group's admin chain, plus the specific
* admin the card was delivered to (named-or-admin);
* - in-flight dedup via the hold's dedup key a retry / rapid second
* message from the same unknown sender is silently dropped (no duplicate
* card), replacing the old sender table's UNIQUE(mg, sender);
* - the hold is sessionless: there is no agent session to notify, so
* failure modes (no approver, no reachable DM, no adapter) log and leave
* no row, letting a future attempt retry.
* 1. Pick an eligible approver (owner / admin of the agent group).
* 2. Open / reuse a DM to that approver on a reachable channel.
* 3. Deliver an Approve / Deny card.
* 4. Record a pending_sender_approvals row that holds the original message
* so it can be re-routed on approve.
*
* On approve: the 'sender_admit' handler in index.ts adds an
* agent_group_members row for the sender and re-invokes routeInbound with the
* stored event the second routing attempt passes the gate because the user
* is now a member. On deny: the shared reject path just drops the hold (no
* denial persistence a future message re-triggers a fresh card).
* On approve: the handler in index.ts adds an agent_group_members row for
* the sender and re-invokes routeInbound with the stored event the second
* routing attempt passes the gate because the user is now a member.
*
* Failure modes (logged + row NOT created, so the dedup gate lets a future
* attempt try again):
* - No eligible approver in user_roles fresh install, no owner yet.
* - Approver has no reachable DM (no user_dms row + channel can't
* openDM) e.g. owner hasn't registered on any channel we're wired to.
* - Delivery adapter missing.
*
* Dedup: `pending_sender_approvals` has UNIQUE(messaging_group_id,
* sender_identity). A retry / rapid second message from the same unknown
* sender is silently dropped (no duplicate card sent).
*/
import type { RawOption } from '../../channels/ask-question.js';
import { normalizeOptions, type RawOption } from '../../channels/ask-question.js';
import { getMessagingGroup } from '../../db/messaging-groups.js';
import { getDeliveryAdapter } from '../../delivery.js';
import { log } from '../../log.js';
import type { InboundEvent } from '../../channels/adapter.js';
import { requestApproval } from '../approvals/primitive.js';
import { pickApprovalDelivery, pickApprover } from '../approvals/primitive.js';
import { createPendingSenderApproval, hasInFlightSenderApproval } from './db/pending-sender-approvals.js';
const APPROVAL_OPTIONS: RawOption[] = [
{ label: 'Allow', selectedLabel: '✅ Allowed', value: 'approve', style: 'primary' },
{ label: 'Deny', selectedLabel: '❌ Denied', value: 'reject', style: 'danger' },
];
export const SENDER_ADMIT_ACTION = 'sender_admit';
export function senderAdmitDedupKey(messagingGroupId: string, senderIdentity: string): string {
return `${SENDER_ADMIT_ACTION}:${messagingGroupId}:${senderIdentity}`;
function generateId(): string {
return `nsa-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
}
export interface RequestSenderApprovalInput {
@@ -48,20 +54,102 @@ export interface RequestSenderApprovalInput {
export async function requestSenderApproval(input: RequestSenderApprovalInput): Promise<void> {
const { messagingGroupId, agentGroupId, senderIdentity, senderName, event } = input;
const originMg = getMessagingGroup(messagingGroupId);
const senderDisplay = senderName && senderName.length > 0 ? senderName : senderIdentity;
const originName = originMg?.name ?? `a ${originMg?.channel_type ?? ''} channel`;
// In-flight dedup: don't spam the admin if the same unknown sender
// retries while a card is already pending.
if (hasInFlightSenderApproval(messagingGroupId, senderIdentity)) {
log.debug('Unknown-sender approval already in flight — dropping retry', {
messagingGroupId,
senderIdentity,
});
return;
}
await requestApproval({
agentGroupId,
agentName: senderDisplay,
action: SENDER_ADMIT_ACTION,
payload: { messagingGroupId, agentGroupId, senderIdentity, senderName, event },
title: '👤 New sender',
question: `${senderDisplay} wants to talk to your agent in ${originName}. Allow?`,
options: APPROVAL_OPTIONS,
dedupKey: senderAdmitDedupKey(messagingGroupId, senderIdentity),
recordDeliveredApprover: true,
originChannelType: originMg?.channel_type ?? '',
const approvers = pickApprover(agentGroupId);
if (approvers.length === 0) {
log.warn('Unknown-sender approval skipped — no owner or admin configured', {
messagingGroupId,
agentGroupId,
senderIdentity,
});
return;
}
const originMg = getMessagingGroup(messagingGroupId);
const originChannelType = originMg?.channel_type ?? '';
const target = await pickApprovalDelivery(approvers, originChannelType);
if (!target) {
log.warn('Unknown-sender approval skipped — no DM channel for any approver', {
messagingGroupId,
agentGroupId,
senderIdentity,
});
return;
}
const approvalId = generateId();
const senderDisplay = senderName && senderName.length > 0 ? senderName : senderIdentity;
const originName = originMg?.name ?? `a ${originChannelType} channel`;
const title = '👤 New sender';
const question = `${senderDisplay} wants to talk to your agent in ${originName}. Allow?`;
const options = normalizeOptions(APPROVAL_OPTIONS);
createPendingSenderApproval({
id: approvalId,
messaging_group_id: messagingGroupId,
agent_group_id: agentGroupId,
sender_identity: senderIdentity,
sender_name: senderName,
original_message: JSON.stringify(event),
approver_user_id: target.userId,
created_at: new Date().toISOString(),
title,
options_json: JSON.stringify(options),
});
const adapter = getDeliveryAdapter();
if (!adapter) {
// Without a delivery adapter, the card can't be sent. Log + leave the
// row in place so the admin can see it via DB or manual tooling; the
// dedup gate will suppress further cards until it's cleared.
log.error('Unknown-sender approval row created but no delivery adapter is wired', {
approvalId,
});
return;
}
try {
await adapter.deliver(
target.messagingGroup.channel_type,
target.messagingGroup.platform_id,
null,
'chat-sdk',
JSON.stringify({
type: 'ask_question',
questionId: approvalId,
title,
question,
options,
}),
);
log.info('Unknown-sender approval card delivered', {
approvalId,
senderIdentity,
approver: target.userId,
messagingGroupId,
agentGroupId,
});
} catch (err) {
log.error('Unknown-sender approval card delivery failed', {
approvalId,
err,
});
}
}
/**
* Option value the admin clicked that means "allow" shared with the
* response handler so the two sides can't drift.
*/
export const APPROVE_VALUE = 'approve';
export const REJECT_VALUE = 'reject';
-113
View File
@@ -1,113 +0,0 @@
/**
* Delivery action handlers for scheduling.
*
* The container can't write to inbound.db (host-owned). When the agent calls
* schedule_task / cancel_task / etc. via MCP, the container writes a
* `kind='system'` outbound message with an `action` field. The delivery path
* reaches into this module via the delivery-action registry and we apply the
* change to inbound.db here.
*/
import type Database from 'better-sqlite3';
import { wakeContainer } from '../../container-runner.js';
import { getSession } from '../../db/sessions.js';
import { log } from '../../log.js';
import { writeSessionMessage } from '../../session-manager.js';
import type { Session } from '../../types.js';
import { cancelTask, insertTask, pauseTask, resumeTask, updateTask, type TaskUpdate } from './db.js';
export async function handleScheduleTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
const prompt = content.prompt as string;
const script = content.script as string | null;
const processAfter = content.processAfter as string;
const recurrence = (content.recurrence as string) || null;
insertTask(inDb, {
id: taskId,
processAfter,
recurrence,
platformId: (content.platformId as string) ?? null,
channelType: (content.channelType as string) ?? null,
threadId: (content.threadId as string) ?? null,
content: JSON.stringify({ prompt, script }),
});
log.info('Scheduled task created', { taskId, processAfter, recurrence });
}
export async function handleCancelTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
cancelTask(inDb, taskId);
log.info('Task cancelled', { taskId });
}
export async function handlePauseTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
pauseTask(inDb, taskId);
log.info('Task paused', { taskId });
}
export async function handleResumeTask(
content: Record<string, unknown>,
_session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
resumeTask(inDb, taskId);
log.info('Task resumed', { taskId });
}
export async function handleUpdateTask(
content: Record<string, unknown>,
session: Session,
inDb: Database.Database,
): Promise<void> {
const taskId = content.taskId as string;
const update: TaskUpdate = {};
if (typeof content.prompt === 'string') update.prompt = content.prompt;
if (typeof content.processAfter === 'string') update.processAfter = content.processAfter;
if (content.recurrence === null || typeof content.recurrence === 'string') {
update.recurrence = content.recurrence as string | null;
}
if (content.script === null || typeof content.script === 'string') {
update.script = content.script as string | null;
}
const touched = updateTask(inDb, taskId, update);
log.info('Task updated', { taskId, touched, fields: Object.keys(update) });
if (touched === 0) {
// Notify the agent that update_task matched nothing. Replicates the
// old notifyAgent helper that used to live in delivery.ts — inlined
// here so scheduling doesn't depend on delivery's private helpers.
writeSessionMessage(session.agent_group_id, session.id, {
id: `sys-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
kind: 'chat',
timestamp: new Date().toISOString(),
platformId: session.agent_group_id,
channelType: 'agent',
threadId: null,
content: JSON.stringify({
text: `update_task: no live task matched id "${taskId}".`,
sender: 'system',
senderId: 'system',
}),
});
const fresh = getSession(session.id);
if (fresh) {
wakeContainer(fresh).catch((err) =>
log.error('Failed to wake container after update_task notification', { err }),
);
}
}
}
+37 -42
View File
@@ -10,7 +10,7 @@ import { describe, it, expect, afterEach } from 'vitest';
import { ensureSchema, openInboundDb } from '../../db/session-db.js';
import {
insertTask,
insertTaskRow,
insertRecurrence,
cancelTask,
pauseTask,
@@ -31,13 +31,11 @@ function freshDb() {
}
function insertBasicTask(db: ReturnType<typeof openInboundDb>, id: string, recurrence: string | null) {
insertTask(db, {
insertTaskRow(db, {
id,
seriesId: id,
processAfter: new Date().toISOString(),
recurrence,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'noop' }),
});
}
@@ -46,7 +44,7 @@ afterEach(() => {
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
});
describe('insertTask', () => {
describe('insertTaskRow', () => {
it('stamps series_id = id on insert', () => {
const db = freshDb();
insertBasicTask(db, 'task-1', null);
@@ -68,13 +66,8 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: JSON.stringify({ prompt: 'noop' }),
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date(Date.now() + 86400000).toISOString());
@@ -93,7 +86,8 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
status: string;
recurrence: string | null;
};
expect(followUp.status).toBe('completed');
// Cancel marks 'cancelled' (not 'completed') so it never counts as a run.
expect(followUp.status).toBe('cancelled');
// Recurrence cleared so the sweep doesn't spawn another clone.
expect(followUp.recurrence).toBeNull();
db.close();
@@ -131,18 +125,37 @@ describe('cancelTask / pauseTask / resumeTask series matching', () => {
expect(followUp.status).toBe('pending');
db.close();
});
it('pause/resume touch ONLY status — recurrence and process_after survive the cycle', () => {
const db = freshDb();
seedRecurringChain(db);
const before = db.prepare("SELECT recurrence, process_after FROM messages_in WHERE id = 'task-next'").get() as {
recurrence: string | null;
process_after: string | null;
};
pauseTask(db, 'task-next');
resumeTask(db, 'task-next');
const after = db.prepare("SELECT recurrence, process_after FROM messages_in WHERE id = 'task-next'").get() as {
recurrence: string | null;
process_after: string | null;
};
// A cancel-style copy-paste (clearing recurrence) would kill the series here.
expect(after.recurrence).toBe(before.recurrence);
expect(after.process_after).toBe(before.process_after);
db.close();
});
});
describe('updateTask', () => {
it('merges supplied fields into content JSON without clobbering others', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: new Date().toISOString(),
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'old', script: 'echo old', extra: 'keep me' }),
});
@@ -158,13 +171,11 @@ describe('updateTask', () => {
it('updates recurrence and process_after when supplied', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2026-01-01T00:00:00Z',
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
@@ -180,13 +191,11 @@ describe('updateTask', () => {
it('clears recurrence when null is passed', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2026-01-01T00:00:00Z',
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
@@ -200,26 +209,19 @@ describe('updateTask', () => {
it('reaches the live follow-up via series_id when called with the original id', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-orig',
seriesId: 'task-orig',
processAfter: new Date().toISOString(),
recurrence: '0 9 * * *',
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'old' }),
});
db.prepare("UPDATE messages_in SET status = 'completed' WHERE id = 'task-orig'").run();
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: JSON.stringify({ prompt: 'old' }),
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date(Date.now() + 86400000).toISOString());
@@ -238,13 +240,11 @@ describe('updateTask', () => {
it('returns 0 when no live task matches', () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: new Date().toISOString(),
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'p' }),
});
db.prepare("UPDATE messages_in SET status = 'completed' WHERE id = 'task-1'").run();
@@ -262,13 +262,8 @@ describe('insertRecurrence', () => {
const msg: RecurringMessage = {
id: 'task-orig',
kind: 'task',
content: '{}',
recurrence: '0 9 * * *',
process_after: null,
platform_id: null,
channel_type: null,
thread_id: null,
series_id: 'task-orig',
};
insertRecurrence(db, msg, 'task-next', new Date().toISOString());
+88 -41
View File
@@ -14,43 +14,71 @@ import type Database from 'better-sqlite3';
import { nextEvenSeq } from '../../db/session-db.js';
export function insertTask(
/**
* Insert one pending task occurrence. `seriesId` is the series join key equal
* to `id` for a brand-new series, or the existing series for a recurrence clone
* or an on-demand run. Tasks never set platform/channel/thread (they fire into
* an isolated system session), so those columns are always NULL.
*/
export function insertTaskRow(
db: Database.Database,
task: {
row: {
id: string;
processAfter: string;
seriesId: string;
processAfter: string | null;
recurrence: string | null;
platformId: string | null;
channelType: string | null;
threadId: string | null;
content: string;
status?: 'pending' | 'paused';
},
): void {
db.prepare(
`INSERT INTO messages_in (id, seq, timestamp, status, tries, process_after, recurrence, kind, platform_id, channel_type, thread_id, content, series_id)
VALUES (@id, @seq, datetime('now'), 'pending', 0, @processAfter, @recurrence, 'task', @platformId, @channelType, @threadId, @content, @id)`,
VALUES (@id, @seq, datetime('now'), @status, 0, @processAfter, @recurrence, 'task', NULL, NULL, NULL, @content, @seriesId)`,
).run({
...task,
status: 'pending',
...row,
seq: nextEvenSeq(db),
});
}
export function cancelTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'completed', recurrence = NULL WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status IN ('pending', 'paused')",
).run(taskId, taskId);
// Cancel marks the live row 'cancelled' (not 'completed') so a never-fired
// occurrence is distinguishable from a real run and never inflates run history;
// recurrence is cleared so the series isn't re-armed by handleRecurrence.
export function cancelTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'cancelled', recurrence = NULL WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status IN ('pending', 'paused')",
)
.run(taskId, taskId).changes;
}
export function pauseTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'paused' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'pending'",
).run(taskId, taskId);
export function cancelAllTasks(db: Database.Database): number {
return db
.prepare(
"UPDATE messages_in SET status = 'cancelled', recurrence = NULL WHERE kind = 'task' AND status IN ('pending', 'paused')",
)
.run().changes;
}
export function resumeTask(db: Database.Database, taskId: string): void {
db.prepare(
"UPDATE messages_in SET status = 'pending' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'paused'",
).run(taskId, taskId);
export function pauseTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'paused' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'pending'",
)
.run(taskId, taskId).changes;
}
export function resumeTask(db: Database.Database, taskId: string): number {
return db
.prepare(
"UPDATE messages_in SET status = 'pending' WHERE (id = ? OR series_id = ?) AND kind = 'task' AND status = 'paused'",
)
.run(taskId, taskId).changes;
}
export function deleteTask(db: Database.Database, taskId: string): number {
return db.prepare("DELETE FROM messages_in WHERE (id = ? OR series_id = ?) AND kind = 'task'").run(taskId, taskId)
.changes;
}
export interface TaskUpdate {
@@ -107,45 +135,64 @@ export function updateTask(db: Database.Database, taskId: string, update: TaskUp
return rows.length;
}
// Only tasks carry a recurrence (non-task writeSessionMessage never sets one),
// so getCompletedRecurring only ever returns task rows — the fields below are
// all that handleRecurrence needs to clone the next occurrence.
export interface RecurringMessage {
id: string;
kind: string;
content: string;
recurrence: string;
process_after: string | null;
platform_id: string | null;
channel_type: string | null;
thread_id: string | null;
series_id: string;
}
// Failed occurrences (script-skip:error runs) re-arm too — a broken monitor
// must keep its series alive so backoff can throttle it and the cap can pause
// it; dropping the row would silently kill the series on first script error.
export function getCompletedRecurring(db: Database.Database): RecurringMessage[] {
return db
.prepare("SELECT * FROM messages_in WHERE status = 'completed' AND recurrence IS NOT NULL")
.prepare("SELECT * FROM messages_in WHERE status IN ('completed', 'failed') AND recurrence IS NOT NULL")
.all() as RecurringMessage[];
}
/**
* Trailing consecutive FAILED occurrences of a series, newest backwards until
* the first completed run. This IS the script-failure streak derived from
* the occurrence history, no stored counter to update or reset. Deliberately
* counts ANY failed occurrence (script-skip:error acks AND stuck-message
* failures from host-sweep's MAX_TRIES path): a series failing for either
* reason should throttle, not spin.
*/
export function trailingFailedRuns(db: Database.Database, seriesKey: string): number {
const rows = db
.prepare(
`SELECT status FROM messages_in
WHERE (series_id = ? OR id = ?) AND kind = 'task' AND status IN ('completed', 'failed')
ORDER BY seq DESC`,
)
.all(seriesKey, seriesKey) as Array<{ status: string }>;
let streak = 0;
for (const r of rows) {
if (r.status !== 'failed') break;
streak++;
}
return streak;
}
export function insertRecurrence(
db: Database.Database,
msg: RecurringMessage,
newId: string,
nextRun: string | null,
status: 'pending' | 'paused' = 'pending',
): void {
db.prepare(
`INSERT INTO messages_in (id, seq, kind, timestamp, status, process_after, recurrence, platform_id, channel_type, thread_id, content, series_id)
VALUES (?, ?, ?, datetime('now'), 'pending', ?, ?, ?, ?, ?, ?, ?)`,
).run(
newId,
nextEvenSeq(db),
msg.kind,
nextRun,
msg.recurrence,
msg.platform_id,
msg.channel_type,
msg.thread_id,
msg.content,
msg.series_id,
);
insertTaskRow(db, {
id: newId,
seriesId: msg.series_id,
processAfter: nextRun,
recurrence: msg.recurrence,
content: msg.content,
status,
});
}
export function clearRecurrence(db: Database.Database, messageId: string): void {
-34
View File
@@ -1,34 +0,0 @@
/**
* Scheduling module one-shot and recurring tasks.
*
* Registers:
* - Five delivery action handlers: schedule_task, cancel_task, pause_task,
* resume_task, update_task. The container's scheduling MCP tools
* (container/agent-runner/src/mcp-tools/scheduling.ts) write system
* messages with these actions; the host applies them to inbound.db.
*
* Host integration points (filled by MODULE-HOOK markers, validated here
* with the scheduling module shipping inline):
* - `src/host-sweep.ts` MODULE-HOOK:scheduling-recurrence calls
* `handleRecurrence` each sweep tick.
* - `container/agent-runner/src/poll-loop.ts` MODULE-HOOK:scheduling-pre-task
* runs `applyPreTaskScripts` before the provider call so tasks carrying
* a pre-agent script can gate their own execution.
*
* No DB migration tasks are `messages_in` rows with `kind='task'`, so the
* module piggybacks on the core schema.
*/
import { registerDeliveryAction } from '../../delivery.js';
import {
handleCancelTask,
handlePauseTask,
handleResumeTask,
handleScheduleTask,
handleUpdateTask,
} from './actions.js';
registerDeliveryAction('schedule_task', handleScheduleTask);
registerDeliveryAction('cancel_task', handleCancelTask);
registerDeliveryAction('pause_task', handlePauseTask);
registerDeliveryAction('resume_task', handleResumeTask);
registerDeliveryAction('update_task', handleUpdateTask);
+106 -11
View File
@@ -8,13 +8,20 @@
*/
import fs from 'fs';
import path from 'path';
import { afterEach, describe, expect, it } from 'vitest';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { ensureSchema, openInboundDb } from '../../db/session-db.js';
import { insertTask } from './db.js';
import { handleRecurrence } from './recurrence.js';
import { insertTaskRow } from './db.js';
import { handleRecurrence, scriptBackoffMinutes } from './recurrence.js';
import type { Session } from '../../types.js';
// Pin a non-UTC zone so the tz-interpretation test is exact even on UTC CI.
// Asia/Tokyo is UTC+9 with no DST: "0 9 * * *" must land at 00:00:00Z sharp.
vi.mock('../../config.js', async (importOriginal) => {
const actual = await importOriginal<typeof import('../../config.js')>();
return { ...actual, TIMEZONE: 'Asia/Tokyo' };
});
const TEST_DIR = '/tmp/nanoclaw-recurrence-test';
const DB_PATH = path.join(TEST_DIR, 'inbound.db');
@@ -45,13 +52,11 @@ afterEach(() => {
describe('handleRecurrence', () => {
it('clones a completed recurring task with a next-run in the future', async () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: '0 9 * * *', // every day at 09:00 (user TZ)
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'daily digest' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-1'`).run();
@@ -77,15 +82,34 @@ describe('handleRecurrence', () => {
expect(new Date(follow.process_after).getTime()).toBeGreaterThan(Date.now());
});
it('interprets the cron expression in TIMEZONE, not UTC (the v1 regression)', async () => {
const db = freshDb();
insertTaskRow(db, {
id: 'task-tz',
seriesId: 'task-tz',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: '0 9 * * *', // 09:00 Asia/Tokyo === 00:00 UTC, exactly
content: JSON.stringify({ prompt: 'daily digest' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-tz'`).run();
await handleRecurrence(db, fakeSession());
const follow = db.prepare(`SELECT process_after FROM messages_in WHERE id != 'task-tz'`).get() as {
process_after: string;
};
// Drop the `{ tz: TIMEZONE }` option in recurrence.ts and this reads
// T09:00:00 (09:00 UTC) instead — red, even on a UTC CI runner.
expect(follow.process_after).toMatch(/T00:00:00/);
});
it('does not clone rows whose recurrence is already cleared', async () => {
const db = freshDb();
insertTask(db, {
insertTaskRow(db, {
id: 'task-1',
seriesId: 'task-1',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: null,
platformId: null,
channelType: null,
threadId: null,
content: JSON.stringify({ prompt: 'one-off' }),
});
db.prepare(`UPDATE messages_in SET status='completed' WHERE id='task-1'`).run();
@@ -96,3 +120,74 @@ describe('handleRecurrence', () => {
expect(count).toBe(1);
});
});
describe('handleRecurrence — script-failure backoff (streak derived from failed runs)', () => {
// A series whose last `fails` occurrences all landed as FAILED (script-skip:error
// runs, as synced by syncProcessingAcks). Only the newest row keeps recurrence —
// older occurrences had theirs cleared when they were re-armed. fails=0 seeds one
// healthy completed run.
function seedFailedStreak(db: ReturnType<typeof freshDb>, fails: number) {
const rows = Math.max(fails, 1);
for (let i = 0; i < rows; i++) {
insertTaskRow(db, {
id: `task-s-${i}`,
seriesId: 'task-s-0',
processAfter: '2020-01-01T00:00:00.000Z',
recurrence: i === rows - 1 ? '* * * * *' : null, // every minute — raw cron next is ~+1min
content: JSON.stringify({ prompt: 'monitor', script: 'exit 1' }),
});
db.prepare(`UPDATE messages_in SET status = ? WHERE id = ?`).run(
fails === 0 ? 'completed' : 'failed',
`task-s-${i}`,
);
}
return `task-s-${rows - 1}`; // the row carrying recurrence
}
const clone = (db: ReturnType<typeof freshDb>) =>
db.prepare(`SELECT status, process_after, recurrence FROM messages_in WHERE id NOT LIKE 'task-s-%'`).get() as {
status: string;
process_after: string;
recurrence: string | null;
};
it('exports the documented 2,4,8,…,60 progression', () => {
expect([1, 2, 3, 4, 5, 6, 7].map(scriptBackoffMinutes)).toEqual([2, 4, 8, 16, 32, 60, 60]);
});
it('pushes the clone past raw cron cadence while the script is failing', async () => {
const db = freshDb();
seedFailedStreak(db, 3); // streak 3 → backoff 8 min; cron next ≈ +1 min
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('pending');
const deltaMin = (new Date(next.process_after).getTime() - Date.now()) / 60_000;
expect(deltaMin).toBeGreaterThan(7); // backoff won over the 1-min cron grid
});
it('a healthy series (trailing run completed) re-arms on the raw cron grid', async () => {
const db = freshDb();
seedFailedStreak(db, 0);
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('pending');
const deltaMin = (new Date(next.process_after).getTime() - Date.now()) / 60_000;
expect(deltaMin).toBeLessThan(2); // no backoff applied
});
it('auto-pauses the series at the cap instead of re-arming', async () => {
const db = freshDb();
const liveId = seedFailedStreak(db, 8);
await handleRecurrence(db, fakeSession());
const next = clone(db);
expect(next.status).toBe('paused'); // `ncl tasks resume` revives in place
expect(next.recurrence).toBe('* * * * *');
const original = db.prepare(`SELECT recurrence FROM messages_in WHERE id = ?`).get(liveId) as {
recurrence: string | null;
};
expect(original.recurrence).toBeNull(); // not re-cloned next sweep
});
});
+60 -7
View File
@@ -11,27 +11,79 @@
* direct dynamic import. When scheduling moves to the modules branch in
* PR #8, the install skill re-fills the marker on install.
*/
import type Database from 'better-sqlite3';
import fs from 'fs';
import path from 'path';
import { TIMEZONE } from '../../config.js';
import type Database from 'better-sqlite3';
import { CronExpressionParser } from 'cron-parser';
import { GROUPS_DIR, TIMEZONE } from '../../config.js';
import { getAgentGroup } from '../../db/agent-groups.js';
import { log } from '../../log.js';
import type { Session } from '../../types.js';
import { clearRecurrence, getCompletedRecurring, insertRecurrence } from './db.js';
import { clearRecurrence, getCompletedRecurring, insertRecurrence, trailingFailedRuns } from './db.js';
// Consecutive pre-task-script failures (the series' trailing FAILED runs —
// derived from occurrence rows, no stored counter) throttle a broken monitor
// script instead of letting it wake a container at raw cron cadence forever.
// A deliberate wakeAgent=false gate is a normal completed run and never backs
// off. Mirrors the stuck-message retry in host-sweep.ts (BACKOFF_BASE_MS
// doubling, MAX_TRIES → failed): fail loud, don't spin.
const SCRIPT_FAIL_PAUSE_CAP = 8;
const SCRIPT_BACKOFF_CAP_MIN = 60;
/** 2, 4, 8, 16, 32, 60, 60… minutes for fails = 1, 2, 3… */
export function scriptBackoffMinutes(fails: number): number {
return Math.min(2 * 2 ** (fails - 1), SCRIPT_BACKOFF_CAP_MIN);
}
/** Host-written line in the series run log no agent session exists to call
* append-log when a script-gated series is auto-paused. Same format as
* appendTaskLog (tasks.ts). */
function appendHostTaskNote(agentGroupId: string, seriesId: string, note: string): void {
const ag = getAgentGroup(agentGroupId);
if (!ag) return;
const dir = path.join(GROUPS_DIR, ag.folder, 'tasks');
const timestamp = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
fs.mkdirSync(dir, { recursive: true });
fs.appendFileSync(path.join(dir, `${seriesId}.md`), `${timestamp}${note}\n`);
}
export async function handleRecurrence(inDb: Database.Database, session: Session): Promise<void> {
const recurring = getCompletedRecurring(inDb);
for (const msg of recurring) {
try {
const { CronExpressionParser } = await import('cron-parser');
// Interpret the cron expression in the user's timezone. v1 did this
// (src/v1/task-scheduler.ts:20-49); without it, a task written "0 9 * * *"
// by an agent running in a user's local TZ fires at 09:00 UTC instead of
// 09:00 user-local.
const interval = CronExpressionParser.parse(msg.recurrence, { tz: TIMEZONE });
const nextRun = interval.next().toISOString();
const prefix = msg.kind === 'task' ? 'task' : 'msg';
const newId = `${prefix}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const cronNext = interval.next().toDate();
const newId = `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const scriptFails = trailingFailedRuns(inDb, msg.series_id ?? msg.id);
if (scriptFails >= SCRIPT_FAIL_PAUSE_CAP) {
// Re-arm PAUSED at the cron time so `ncl tasks resume` revives the
// series in place; leave the why in the run log.
insertRecurrence(inDb, msg, newId, cronNext.toISOString(), 'paused');
clearRecurrence(inDb, msg.id);
appendHostTaskNote(
session.agent_group_id,
msg.series_id,
`auto-paused after ${scriptFails} consecutive script failures (host); fix the script, then \`ncl tasks resume ${msg.series_id}\``,
);
log.warn('Task series auto-paused: script keeps failing', {
seriesId: msg.series_id,
scriptFails,
sessionId: session.id,
});
continue;
}
const backoffAt = scriptFails > 0 ? Date.now() + scriptBackoffMinutes(scriptFails) * 60_000 : 0;
const nextRun = new Date(Math.max(cronNext.getTime(), backoffAt)).toISOString();
insertRecurrence(inDb, msg, newId, nextRun);
clearRecurrence(inDb, msg.id);
@@ -41,6 +93,7 @@ export async function handleRecurrence(inDb: Database.Database, session: Session
newId,
seriesId: msg.series_id,
nextRun,
...(scriptFails > 0 && { scriptFails, backoffMin: scriptBackoffMinutes(scriptFails) }),
sessionId: session.id,
});
} catch (err) {
-8
View File
@@ -20,10 +20,6 @@ import { writeSessionMessage } from '../../session-manager.js';
import type { ApprovalHandler } from '../approvals/index.js';
export const applyInstallPackages: ApprovalHandler = async ({ session, payload, userId, notify }) => {
if (!session) {
log.warn('install_packages approval resolved without a session — dropping');
return;
}
const agentGroup = getAgentGroup(session.agent_group_id);
if (!agentGroup) {
notify('install_packages approved but agent group missing.');
@@ -87,10 +83,6 @@ export const applyInstallPackages: ApprovalHandler = async ({ session, payload,
};
export const applyAddMcpServer: ApprovalHandler = async ({ session, payload, userId, notify }) => {
if (!session) {
log.warn('add_mcp_server approval resolved without a session — dropping');
return;
}
const agentGroup = getAgentGroup(session.agent_group_id);
if (!agentGroup) {
notify('add_mcp_server approved but agent group missing.');
+39
View File
@@ -22,9 +22,11 @@ import { ensureContainedInboxDir, isPathInside } from './inbox-safety.js';
import { getMessagingGroup } from './db/messaging-groups.js';
import {
createSession,
findSystemSession,
findSessionByAgentGroup,
findSessionForAgent,
getSession,
taskThreadId,
updateSession,
} from './db/sessions.js';
import {
@@ -120,6 +122,33 @@ export function resolveSession(
return { session, created: true };
}
/** Find or create the per-agent-group session used for scheduled tasks. */
/** Find or create the isolated session for one task series (thread `system:tasks:<seriesId>`). */
export function resolveTaskSession(agentGroupId: string, seriesId: string): { session: Session; created: boolean } {
const threadId = taskThreadId(seriesId);
const existing = findSystemSession(agentGroupId, threadId);
if (existing) return { session: existing, created: false };
const id = generateId();
const session: Session = {
id,
agent_group_id: agentGroupId,
messaging_group_id: null,
thread_id: threadId,
agent_provider: null,
status: 'active',
container_status: 'stopped',
last_active: null,
created_at: new Date().toISOString(),
};
createSession(session);
initSessionFolder(agentGroupId, id);
log.info('Task session created', { id, agentGroupId, seriesId });
return { session, created: true };
}
/** Create the session folder and initialize both DBs. */
export function initSessionFolder(agentGroupId: string, sessionId: string): void {
const dir = sessionDir(agentGroupId, sessionId);
@@ -350,6 +379,16 @@ export function openInboundDb(agentGroupId: string, sessionId: string): Database
return db;
}
/** Open a session's inbound DB, run `fn`, and always close it. */
export function withInboundDb<T>(agentGroupId: string, sessionId: string, fn: (db: Database.Database) => T): T {
const db = openInboundDb(agentGroupId, sessionId);
try {
return fn(db);
} finally {
db.close();
}
}
/** Open the outbound DB for a session (host reads only). */
export function openOutboundDb(agentGroupId: string, sessionId: string): Database.Database {
return openOutboundDbRaw(outboundDbPath(agentGroupId, sessionId));
+28 -1
View File
@@ -1,6 +1,6 @@
import { describe, it, expect } from 'vitest';
import { formatLocalTime, isValidTimezone, resolveTimezone } from './timezone.js';
import { formatLocalTime, isValidTimezone, parseZonedToUtc, resolveTimezone } from './timezone.js';
// --- formatLocalTime ---
@@ -62,3 +62,30 @@ describe('resolveTimezone', () => {
expect(resolveTimezone('')).toBe('UTC');
});
});
describe('parseZonedToUtc', () => {
const iso = (s: string, tz: string): string => parseZonedToUtc(s, tz).toISOString();
it('reads a naive timestamp as wall-clock in a fixed-offset zone', () => {
expect(iso('2026-06-20T09:00:00', 'Asia/Tokyo')).toBe('2026-06-20T00:00:00.000Z'); // UTC+9
});
it('applies the correct seasonal offset (DST honored)', () => {
// Same wall-clock time, different UTC offset by season — proves the offset
// is computed against the zone's rules, not a fixed guess.
expect(iso('2026-07-01T12:00:00', 'America/New_York')).toBe('2026-07-01T16:00:00.000Z'); // EDT -4
expect(iso('2026-01-01T12:00:00', 'America/New_York')).toBe('2026-01-01T17:00:00.000Z'); // EST -5
});
it('passes a trailing-Z timestamp through unchanged', () => {
expect(iso('2026-06-20T09:00:00Z', 'Asia/Tokyo')).toBe('2026-06-20T09:00:00.000Z');
});
it('passes an explicit offset through', () => {
expect(iso('2026-06-20T09:00:00+02:00', 'Asia/Tokyo')).toBe('2026-06-20T07:00:00.000Z');
});
it('falls back to UTC for an invalid zone', () => {
expect(iso('2026-06-20T09:00:00', 'Not/AZone')).toBe('2026-06-20T09:00:00.000Z');
});
});
+42
View File
@@ -35,3 +35,45 @@ export function formatLocalTime(utcIso: string, timezone: string): string {
hour12: true,
});
}
/**
* Interpret a naive ISO-like timestamp (no trailing `Z`, no offset) as wall-clock
* time in `tz` and return the corresponding UTC Date. Strings that already carry
* offset info (`Z` or `+-HH:MM`) are passed through to the Date constructor.
*/
export function parseZonedToUtc(input: string, tz: string): Date {
const hasOffset = /Z$|[+-]\d{2}:?\d{2}$/.test(input.trim());
if (hasOffset) return new Date(input);
const zone = resolveTimezone(tz);
const asIfUtc = new Date(input + 'Z');
if (Number.isNaN(asIfUtc.getTime())) return asIfUtc;
const fmt = new Intl.DateTimeFormat('en-US', {
timeZone: zone,
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hour12: false,
});
const parts = Object.fromEntries(
fmt
.formatToParts(asIfUtc)
.filter((p) => p.type !== 'literal')
.map((p) => [p.type, p.value]),
);
const hour = parts.hour === '24' ? '00' : parts.hour;
const zonedAsUtcMs = Date.UTC(
Number(parts.year),
Number(parts.month) - 1,
Number(parts.day),
Number(hour),
Number(parts.minute),
Number(parts.second),
);
const offsetMs = zonedAsUtcMs - asIfUtc.getTime();
return new Date(asIfUtc.getTime() - offsetMs);
}
+2 -26
View File
@@ -142,7 +142,7 @@ export interface Session {
// ── Session DB entities ──
export type MessageInKind = 'chat' | 'chat-sdk' | 'task' | 'webhook' | 'system';
export type MessageInStatus = 'pending' | 'processing' | 'completed' | 'failed';
export type MessageInStatus = 'pending' | 'processing' | 'completed' | 'failed' | 'cancelled';
export interface MessageIn {
id: string;
@@ -189,20 +189,6 @@ export interface PendingQuestion {
// ── Pending approvals (central DB) ──
/**
* Who may resolve a hold. `exclusive`: only the named user (an a2a policy's
* approver). `admins-of-scope`: the admin chain of the anchoring agent group
* (owners + global admins when the anchor is null), plus the user the card
* was delivered to when recorded. Evaluation lives in
* src/modules/approvals/approver-rule.ts (`mayResolve`).
*/
export type ApproverRule =
| { kind: 'exclusive'; approverUserId: string }
| { kind: 'admins-of-scope'; agentGroupId: string | null; deliveredTo: string | null };
/** Blast radius of a held action: 'global' holds require an owner or global admin to resolve. */
export type ApproverScope = 'group' | 'global';
export interface PendingApproval {
approval_id: string;
session_id: string | null;
@@ -223,18 +209,8 @@ export interface PendingApproval {
status: 'pending' | 'approved' | 'rejected' | 'expired' | 'awaiting_reason';
title: string;
options_json: string;
/**
* Named approver. Under `approver_rule: 'exclusive'` only this exact user
* may resolve the approval; under 'admins-of-scope' it records the user the
* card was delivered to, who may resolve alongside the scope's admins.
*/
/** When set, only this exact user may resolve the approval. */
approver_user_id: string | null;
/** Who may resolve this hold — see modules/approvals/approver-rule.ts. */
approver_rule: 'exclusive' | 'admins-of-scope';
/** Blast radius: 'global' holds require an owner or global admin to resolve. */
approver_scope: 'group' | 'global';
/** In-flight dedup key: while a row carries this key, a repeat request with the same key is dropped. */
dedup_key: string | null;
}
// ── Agent destinations (central DB) ──