From 8357b1ca620ca351c35c17245c07b2f7ca8610d1 Mon Sep 17 00:00:00 2001 From: Omri Maya Date: Wed, 8 Jul 2026 12:44:35 +0300 Subject: [PATCH 1/7] =?UTF-8?q?feat(tasks):=20one-door=20delivery=20?= =?UTF-8?q?=E2=80=94=20send=5Fmessage=20is=20the=20only=20path=20out=20of?= =?UTF-8?q?=20a=20task=20session?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A task fire has no chat attached, so delivery gets exactly one door: the send_message tool with an explicit destination. Final-text blocks are inert in task sessions; the fire's final text is auto-appended to the series run log (tasks/.md) instead — exactly once per run (skipped when the agent already wrote its own append-log line that fire). Inert blocks are stripped from auto-appended lines and logged as [undelivered → …] so a misaddressed send is visible in the log rather than silently gone. A same-turn nudge catches an agent that ends a fire with inert blocks; the previous echo-suppression heuristics are removed with the ambiguity they served. Chat sessions are unchanged. The host now stamps is_task on session_routing at session creation — the container reads that instead of sniffing the thread-id prefix. BREAKING: task prompts that relied on final-text delivery must call send_message(to=...); tasks created with the standard scaffold already instruct this. Rebuild the container image and restart so runners pick up the new poll loop (see CHANGELOG migration note). Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 3 +- container/agent-runner/src/db/messages-out.ts | 67 ++++- .../agent-runner/src/db/session-routing.ts | 17 +- container/agent-runner/src/formatter.ts | 7 +- .../src/mcp-tools/cli.instructions.md | 2 +- container/agent-runner/src/mcp-tools/core.ts | 17 +- .../src/mcp-tools/scheduling.instructions.md | 2 +- container/agent-runner/src/poll-loop.ts | 112 +++++-- .../agent-runner/src/task-delivery.test.ts | 283 ++++++++++++++++++ src/cli/resources/tasks.ts | 29 +- src/db/schema.ts | 6 +- src/db/session-db.test.ts | 51 +++- src/db/session-db.ts | 22 +- src/delivery.test.ts | 36 ++- src/delivery.ts | 27 +- src/host-core.test.ts | 27 ++ src/modules/scheduling/recurrence.test.ts | 22 +- src/modules/scheduling/recurrence.ts | 24 +- src/modules/scheduling/run-log.ts | 32 ++ src/session-manager.ts | 5 + 20 files changed, 704 insertions(+), 87 deletions(-) create mode 100644 container/agent-runner/src/task-delivery.test.ts create mode 100644 src/modules/scheduling/run-log.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 29565d034a2..71fd4bfec99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,8 @@ All notable changes to NanoClaw will be documented in this file. ## [Unreleased] -- **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n−1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. Also fixed: an explicitly-addressed `` 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] **One-door delivery in task sessions.** A task fire delivers ONLY via the `send_message` tool, and `to` is required there (no default destination). Final-text `` blocks are inert in task sessions, and the fire's final text is auto-appended to the series run log (`tasks/.md`) unless the agent already ran `ncl tasks append-log` that fire — a run is logged exactly once, and the double-send/zero-send ambiguity of having two delivery paths is gone (the previous echo-suppression heuristics are removed). Chat sessions are unchanged. **Migration:** rebuild the agent container image (`./container/build.sh`) and restart so containers pick up the new runner; task prompts that relied on `` blocks must switch to `send_message(to=...)` — existing tasks created with the standard scaffold already instruct this. +- **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n−1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. - [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-` 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-` skill to pull the matching `4.29.0` adapter. diff --git a/container/agent-runner/src/db/messages-out.ts b/container/agent-runner/src/db/messages-out.ts index 1b2d9f8659a..9568e8b3c2f 100644 --- a/container/agent-runner/src/db/messages-out.ts +++ b/container/agent-runner/src/db/messages-out.ts @@ -142,21 +142,62 @@ export function getUndeliveredMessages(): MessageOutRow[] { .all() as MessageOutRow[]; } +/** Highest seq across both session DBs — marks "now" for since-queries. */ +export function maxSeq(): number { + const maxOut = (getOutboundDb().prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_out').get() as { m: number }) + .m; + const maxIn = (getInboundDb().prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_in').get() as { m: number }).m; + return Math.max(maxOut, maxIn); +} + /** - * 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 echo of a send the - * agent already made — the dedup happens where the duplication originates. + * True if the agent invoked `ncl tasks append-log` after `sinceSeq` (the ncl + * binary writes each CLI call as a system cli_request row in messages_out). + * Guards the task-fire auto-log: an explicit append-log this fire suppresses + * the runner's final-text auto-append, so a run is logged exactly once. */ -export function hasIdenticalSend(platformId: string, channelType: string, text: string): boolean { - const row = getOutboundDb() +export function hasAppendLogRequestSince(sinceSeq: number): boolean { + // LIKE, not equality: a positional invocation (`ncl tasks append-log "..."`) + // arrives dash-joined as 'tasks-append-log-' — the host dispatcher + // resolves the tail as the positional id. + const requests = 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`, + `SELECT content FROM messages_out + WHERE seq > $since AND kind = 'system' + AND json_extract(content, '$.action') = 'cli_request' + AND json_extract(content, '$.command') LIKE 'tasks-append-log%'`, ) - .get({ $platform_id: platformId, $channel_type: channelType, $text: text }); - return row != null; + .all({ $since: sinceSeq }) as Array<{ content: string }>; + if (requests.length === 0) return false; + + // Only a request that SUCCEEDED counts as "this fire was logged". Correlate + // each request with its host response in inbound messages_in (content carries + // the requestId + response frame). A request with no readable response yet + // still suppresses — the common success case is the response landing after + // the turn ends, and double-logging is worse than a rare missed line. Only a + // DEFINITIVE failure (frame ok:false) does not suppress. + const inbound = getInboundDb(); + for (const { content } of requests) { + let requestId: string | null = null; + try { + requestId = (JSON.parse(content) as { requestId?: string }).requestId ?? null; + } catch { + // Malformed request row — can't correlate; suppress conservatively. + return true; + } + if (!requestId) return true; + + const resp = inbound + .prepare('SELECT content FROM messages_in WHERE content LIKE $pat ORDER BY seq DESC LIMIT 1') + .get({ $pat: `%"requestId":"${requestId}"%` }) as { content: string } | undefined; + if (!resp) return true; // pending response → treat as logged + try { + const frame = (JSON.parse(resp.content) as { frame?: { ok?: boolean } }).frame; + if (frame?.ok !== false) return true; // ok:true (or unreadable frame) → logged + } catch { + return true; // unparseable response — suppress conservatively + } + // frame.ok === false → this request definitively failed; check the next one. + } + return false; } diff --git a/container/agent-runner/src/db/session-routing.ts b/container/agent-runner/src/db/session-routing.ts index 94abca63100..d5d24633191 100644 --- a/container/agent-runner/src/db/session-routing.ts +++ b/container/agent-runner/src/db/session-routing.ts @@ -14,17 +14,28 @@ export interface SessionRouting { channel_type: string | null; platform_id: string | null; thread_id: string | null; + /** 1 = host stamped this session as a task-series session. */ + is_task: 0 | 1; } export function getSessionRouting(): SessionRouting { const db = getInboundDb(); try { const row = db - .prepare('SELECT channel_type, platform_id, thread_id FROM session_routing WHERE id = 1') + .prepare('SELECT channel_type, platform_id, thread_id, is_task FROM session_routing WHERE id = 1') .get() as SessionRouting | undefined; if (row) return row; } catch { - // Table may not exist on an older session DB — fall through to defaults + // is_task column may not exist yet (host predates the stamp) — retry + // without it and derive task-ness from the thread prefix. + try { + const row = db + .prepare('SELECT channel_type, platform_id, thread_id FROM session_routing WHERE id = 1') + .get() as Omit | undefined; + if (row) return { ...row, is_task: row.thread_id?.startsWith('system:tasks') ? 1 : 0 }; + } catch { + // Table may not exist on an older session DB — fall through to defaults + } } - return { channel_type: null, platform_id: null, thread_id: null }; + return { channel_type: null, platform_id: null, thread_id: null, is_task: 0 }; } diff --git a/container/agent-runner/src/formatter.ts b/container/agent-runner/src/formatter.ts index 35a1f8f88df..a8375f5144d 100644 --- a/container/agent-runner/src/formatter.ts +++ b/container/agent-runner/src/formatter.ts @@ -98,10 +98,9 @@ export interface RoutingContext { channelType: string | null; threadId: string | null; inReplyTo: string | null; - /** Batch is a task fire — explicit `` 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. */ + /** Batch is a task fire. One-door delivery: only the send_message tool + * delivers from a task session; final-text `` blocks are inert + * and the final text auto-appends to the series run log. */ taskFire: boolean; } diff --git a/container/agent-runner/src/mcp-tools/cli.instructions.md b/container/agent-runner/src/mcp-tools/cli.instructions.md index f6673bd909c..40b98671215 100644 --- a/container/agent-runner/src/mcp-tools/cli.instructions.md +++ b/container/agent-runner/src/mcp-tools/cli.instructions.md @@ -34,7 +34,7 @@ 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 ` 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). +- **Scheduling work** — `ncl tasks create`, then `ncl tasks list/get/update/cancel/pause/resume/delete`; `ncl tasks run ` fires one extra run now (testing) without changing the schedule. Each task run auto-logs its final text to the run log; `ncl tasks append-log --msg "…"` is for extra mid-run notes (host-timestamped, not a message). - **Answering questions about the system** — query `ncl` rather than guessing. ### Access rules diff --git a/container/agent-runner/src/mcp-tools/core.ts b/container/agent-runner/src/mcp-tools/core.ts index b9b46e087e1..7d50b86794e 100644 --- a/container/agent-runner/src/mcp-tools/core.ts +++ b/container/agent-runner/src/mcp-tools/core.ts @@ -53,8 +53,20 @@ function resolveRouting( to: string | undefined, ): { channel_type: string; platform_id: string; thread_id: string | null; resolvedName: string } | { error: string } { if (!to) { - // Default: reply to whatever thread/channel this session is bound to. const session = getSessionRouting(); + // Task sessions have no origin chat — "reply in place" is meaningless and + // the single-destination shortcut below would silently pick an arbitrary + // target. One-door rule: a task fire must name its destination. + // Driven by the host-stamped is_task flag (session_routing); the thread + // prefix check is only a fallback for hosts that predate the stamp. + // (Prefix mirrors TASKS_SYSTEM_THREAD_ID on the host, src/db/sessions.ts — + // the two runtimes share no modules.) + if (session.is_task === 1 || session.thread_id?.startsWith('system:tasks')) { + return { + error: `This is a task session — pass to= explicitly. Options: ${destinationList()}`, + }; + } + // Default: reply to whatever thread/channel this session is bound to. if (session.channel_type && session.platform_id) { return { channel_type: session.channel_type, @@ -95,7 +107,8 @@ function resolveRouting( export const sendMessage: McpToolDefinition = { tool: { name: 'send_message', - description: 'Send a message to a named destination. If you have only one destination, you can omit `to`.', + description: + 'Send a message to a named destination. If you have only one destination, you can omit `to` — except in task sessions, where `to` is always required (this tool is the ONLY delivery path there; final text is never sent).', inputSchema: { type: 'object' as const, properties: { diff --git a/container/agent-runner/src/mcp-tools/scheduling.instructions.md b/container/agent-runner/src/mcp-tools/scheduling.instructions.md index 386ceae8231..fcb70a72e96 100644 --- a/container/agent-runner/src/mcp-tools/scheduling.instructions.md +++ b/container/agent-runner/src/mcp-tools/scheduling.instructions.md @@ -1,6 +1,6 @@ ## Task scheduling (`ncl tasks`) -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 `...` or `send_message({ to: "name", ... })`. +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. When it fires, the ONLY way to message anyone is `send_message({ to: "name", ... })` with an explicit destination — final text and `` blocks are not delivered from task sessions; the fire's final text is recorded in the task's run log instead. Pass `--name ""` on create to get a readable task id (e.g. `--name "sales briefing"` → `sales-briefing-a25c`); without it ids are `t-`. diff --git a/container/agent-runner/src/poll-loop.ts b/container/agent-runner/src/poll-loop.ts index 347ad8f3c42..e0acbce988b 100644 --- a/container/agent-runner/src/poll-loop.ts +++ b/container/agent-runner/src/poll-loop.ts @@ -1,6 +1,6 @@ import { findByName, getAllDestinations, type DestinationEntry } from './destinations.js'; import { getPendingMessages, markProcessing, markCompleted, markScriptSkipped, type MessageInRow } from './db/messages-in.js'; -import { hasIdenticalSend, writeMessageOut } from './db/messages-out.js'; +import { hasAppendLogRequestSince, maxSeq, writeMessageOut } from './db/messages-out.js'; import { getInboundDb, touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js'; import { clearContinuation, @@ -340,6 +340,9 @@ export async function processQuery( let queryContinuation: string | undefined; let done = false; let unwrappedNudged = false; + // Once-per-turn guard for the task-fire " block was not delivered" + // nudge — mirrors unwrappedNudged for chat turns. + let taskBlockNudged = false; // Prompt queue for the exchange hook — each result event consumes the // oldest unanswered prompt, except a wrapping-retry result, which answers // the same prompt again. Unused (and unmaintained) when the provider @@ -423,6 +426,7 @@ export async function processQuery( const prompt = formatMessages(keep); log(`Pushing ${keep.length} follow-up message(s) into active query`); unwrappedNudged = false; + taskBlockNudged = false; query.push(prompt); archivePrompts.push(prompt); markCompleted(keptIds); @@ -464,6 +468,10 @@ export async function processQuery( })(); }, ACTIVE_POLL_INTERVAL_MS); + // Seq watermark before the agent runs — anything after this is "this fire" + // for the append-log exactly-once guard. + const turnStartSeq = maxSeq(); + try { for await (const event of query.events) { handleEvent(event, routing); @@ -487,8 +495,15 @@ export async function processQuery( // at all — either way the turn is finished. markCompleted(initialBatchIds); if (event.text) { - const { sent, hasUnwrapped } = dispatchResultText(event.text, routing); - if (sent === 0 && event.isError === true) { + const { sent, hasUnwrapped, taskBlocks } = dispatchResultText(event.text, routing); + const willRetryTaskBlocks = shouldNudgeTaskBlocks(routing.taskFire === true, taskBlocks, taskBlockNudged); + // One-door task delivery: the final text becomes the run log entry + // (guarded — an explicit append-log this fire wins). Errors included: + // a failed fire's text belongs in the run log, not in a chat. + // When we nudge on inert blocks, DEFER the append to the + // retry's result so the fire logs exactly once. + if (routing.taskFire && !willRetryTaskBlocks) autoAppendTaskLog(event.text, turnStartSeq); + if (sent === 0 && event.isError === true && !routing.taskFire) { // Non-retryable error turn (e.g. a 403 billing_error) with no // envelope: deliver the notice instead of dropping it as // scratchpad, and skip the re-wrap nudge — it would just re-hammer @@ -520,9 +535,20 @@ export async function processQuery( `Please re-send your response with the correct wrapping.`, ); } - // The wrapping-retry result answers the SAME user prompt — keep it - // queued so the retry archives against it, not the nudge text. - if (!willRetryWrapping) archivePrompts.shift(); + if (willRetryTaskBlocks) { + taskBlockNudged = true; + const names = getAllDestinations() + .map((d) => d.name) + .join(', '); + query.push( + `Your block was NOT delivered — task sessions deliver only via the send_message tool. ` + + `Re-send now with send_message({to: "", ...}). Your destinations: ${names}.`, + ); + } + // A retry result (wrapping or task-block nudge) answers the SAME + // user prompt — keep it queued so the retry archives against it, + // not the nudge text. + if (!willRetryWrapping && !willRetryTaskBlocks) archivePrompts.shift(); } } else { archivePrompts.shift(); @@ -605,11 +631,17 @@ function deliverErrorResult(text: string, routing: RoutingContext): void { * The agent must always wrap output in ... * blocks, even with a single destination. Bare text is scratchpad only. */ -function dispatchResultText(text: string, routing: RoutingContext): { sent: number; hasUnwrapped: boolean } { +export function dispatchResultText( + text: string, + routing: RoutingContext, +): { sent: number; hasUnwrapped: boolean; taskBlocks: number } { const MESSAGE_RE = /([\s\S]*?)<\/message>/g; let match: RegExpExecArray | null; let sent = 0; + // blocks left inert in a task fire — drives the same-turn + // "use send_message" nudge in processQuery. + let taskBlocks = 0; let lastIndex = 0; const scratchpadParts: string[] = []; @@ -621,6 +653,16 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb const body = match[2].trim(); lastIndex = MESSAGE_RE.lastIndex; + // One-door delivery in task sessions: only the send_message tool delivers. + // A final-text block here is either an echo of a tool send the + // agent already made (the double-delivery class) or a send down the wrong + // path — never deliver it, keep it visible in the scratchpad/run log. + if (routing.taskFire) { + log(`Task fire: block not delivered — task sessions send only via send_message`); + scratchpadParts.push(`[not delivered — task sessions send only via the send_message tool; to="${toName}"] ${body}`); + taskBlocks++; + continue; + } const dest = findByName(toName); if (!dest) { log(`Unknown destination in , dropping block`); @@ -640,25 +682,57 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb log(`[scratchpad] ${scratchpad.slice(0, 500)}${scratchpad.length > 500 ? '…' : ''}`); } - const hasUnwrapped = sent === 0 && !!scratchpad; + // In a task fire, plain final text is the NORMAL ending (it becomes the run + // log) — never treat it as an undelivered reply or nudge the agent to wrap it. + const hasUnwrapped = !routing.taskFire && sent === 0 && !!scratchpad; if (hasUnwrapped) { log(`WARNING: agent output had no blocks — nothing was sent`); } - return { sent, hasUnwrapped }; + return { sent, hasUnwrapped, taskBlocks }; +} + +/** + * Should this task-fire result get the same-turn "your block was + * not delivered — use send_message" nudge? True at most once per turn + * (mirrors the unwrappedNudged flag for chat turns). While true, the run-log + * auto-append is DEFERRED to the retry's result so the fire logs exactly once. + */ +export function shouldNudgeTaskBlocks(taskFire: boolean, taskBlocks: number, alreadyNudged: boolean): boolean { + return taskFire && taskBlocks > 0 && !alreadyNudged; +} + +/** + * Task fires: the final text IS the run log entry, unless the agent already + * logged this fire explicitly via `ncl tasks append-log` (exactly-once guard — + * old tasks whose baked-in prompt still mandates append-log don't double-log). + * Written as a `task_log` outbound row; the host appends it to the series' + * tasks/.md with its usual timestamp stamp. Never delivered to anyone. + */ +export function autoAppendTaskLog(text: string, turnStartSeq: number): void { + // Run-log hygiene: an inert block never belongs in the log as + // raw XML — replace each with its inner text, marked undelivered, so the + // log stays readable prose. + const prose = text.replace( + /([\s\S]*?)<\/message>/g, + (_m, to: string, body: string) => `[undelivered → ${to}] ${body.trim()}`, + ); + const line = stripInternalTags(prose).replace(/\s+/g, ' ').trim().slice(0, 500); + if (!line) return; + if (hasAppendLogRequestSince(turnStartSeq)) { + log('Task fire already logged via append-log — skipping final-text auto-log'); + return; + } + writeMessageOut({ + id: generateId(), + kind: 'task_log', + content: JSON.stringify({ text: line }), + }); + log('Task fire run log auto-appended from final text'); } 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 @@ -666,7 +740,7 @@ function sendToDestination(dest: DestinationEntry, body: string, routing: Routin const destRouting = resolveDestinationThread(channelType, platformId); writeMessageOut({ id: generateId(), - in_reply_to: destRouting?.inReplyTo ?? (routing.taskFire ? null : routing.inReplyTo), + in_reply_to: destRouting?.inReplyTo ?? routing.inReplyTo, kind: 'chat', platform_id: platformId, channel_type: channelType, diff --git a/container/agent-runner/src/task-delivery.test.ts b/container/agent-runner/src/task-delivery.test.ts new file mode 100644 index 00000000000..63e1e4145fb --- /dev/null +++ b/container/agent-runner/src/task-delivery.test.ts @@ -0,0 +1,283 @@ +/** + * One-door delivery in task sessions. + * + * Wiring under test (each leg goes red if its integration is deleted): + * 1. send_message with no `to` ERRORS in a task session (session_routing + * thread system:tasks:*) instead of falling back to a default target. + * 2. Final-text `` blocks are inert in a task fire — no + * outbound chat row, no "undelivered" nudge state. + * 3. The fire's final text auto-appends as a `task_log` outbound row, + * EXCEPT when the agent already ran `ncl tasks append-log` this fire. + */ +import { describe, it, expect, beforeEach, afterEach } from 'bun:test'; + +import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from './db/connection.js'; +import { getUndeliveredMessages, hasAppendLogRequestSince, maxSeq, writeMessageOut } from './db/messages-out.js'; +import { sendMessage } from './mcp-tools/core.js'; +import { autoAppendTaskLog, dispatchResultText, shouldNudgeTaskBlocks } from './poll-loop.js'; +import type { RoutingContext } from './formatter.js'; + +function seedSessionRouting(threadId: string | null, isTask?: 0 | 1): void { + const db = getInboundDb(); + db.exec(`CREATE TABLE IF NOT EXISTS session_routing ( + id INTEGER PRIMARY KEY CHECK (id = 1), + channel_type TEXT, platform_id TEXT, thread_id TEXT, + is_task INTEGER NOT NULL DEFAULT 0 + )`); + db.prepare( + 'INSERT OR REPLACE INTO session_routing (id, channel_type, platform_id, thread_id, is_task) VALUES (1, ?, ?, ?, ?)', + ).run( + threadId ? null : 'telegram', + threadId ? null : 'telegram:123', + threadId, + isTask ?? (threadId?.startsWith('system:tasks') ? 1 : 0), + ); +} + +function seedDestination(): void { + getInboundDb() + .prepare( + `INSERT INTO destinations (name, display_name, type, channel_type, platform_id, agent_group_id) + VALUES ('family', 'Family', 'channel', 'telegram', 'telegram:99', NULL)`, + ) + .run(); +} + +const taskRouting: RoutingContext = { + platformId: 'ag-1', + channelType: 'agent', + threadId: 'system:tasks:daily-digest-a1b2', + inReplyTo: 'fire-1', + taskFire: true, +}; + +beforeEach(() => { + initTestSessionDb(); + seedDestination(); +}); + +afterEach(() => { + closeSessionDb(); +}); + +describe('send_message in a task session', () => { + it('errors without `to` — no default-destination fallback', async () => { + seedSessionRouting('system:tasks:daily-digest-a1b2'); + + const res = (await sendMessage.handler({ text: 'hello' })) as { isError?: boolean; content: { text: string }[] }; + + expect(res.isError).toBe(true); + expect(res.content[0].text).toContain('task session'); + expect(getUndeliveredMessages()).toHaveLength(0); + }); + + it('the host-stamped is_task flag alone drives the gate (thread_id NULL)', async () => { + // No thread prefix to sniff — proves the flag, not the magic string, + // is what makes the session a task session. + seedSessionRouting(null, 1); + + const res = (await sendMessage.handler({ text: 'hello' })) as { isError?: boolean; content: { text: string }[] }; + + expect(res.isError).toBe(true); + expect(res.content[0].text).toContain('task session'); + expect(getUndeliveredMessages()).toHaveLength(0); + }); + + it('delivers normally with an explicit `to`', async () => { + seedSessionRouting('system:tasks:daily-digest-a1b2'); + + await sendMessage.handler({ to: 'family', text: 'hello' }); + + const out = getUndeliveredMessages(); + expect(out).toHaveLength(1); + expect(out[0].platform_id).toBe('telegram:99'); + }); + + it('chat sessions keep the reply-in-place default', async () => { + seedSessionRouting(null); // normal chat routing row + + await sendMessage.handler({ text: 'hello' }); + + const out = getUndeliveredMessages(); + expect(out).toHaveLength(1); + expect(out[0].platform_id).toBe('telegram:123'); + }); +}); + +describe('final-text blocks in a task fire', () => { + it('are inert — no outbound row, no undelivered flag, counted for the nudge', () => { + const { sent, hasUnwrapped, taskBlocks } = dispatchResultText( + 'digest is ready', + taskRouting, + ); + + expect(sent).toBe(0); + expect(hasUnwrapped).toBe(false); // plain text is the normal task ending — never the wrap nudge + expect(taskBlocks).toBe(1); // but the inert block IS flagged for the task nudge + expect(getUndeliveredMessages()).toHaveLength(0); + }); + + it('still deliver in non-task sessions', () => { + const { sent, taskBlocks } = dispatchResultText('hi', { + ...taskRouting, + taskFire: false, + }); + + expect(sent).toBe(1); + expect(taskBlocks).toBe(0); + expect(getUndeliveredMessages()).toHaveLength(1); + }); + + it('nudge fires once per turn, and only in task fires with blocks', () => { + expect(shouldNudgeTaskBlocks(true, 1, false)).toBe(true); + expect(shouldNudgeTaskBlocks(true, 1, true)).toBe(false); // already nudged this turn + expect(shouldNudgeTaskBlocks(true, 0, false)).toBe(false); // plain text is the normal ending + expect(shouldNudgeTaskBlocks(false, 1, false)).toBe(false); // chat turns use the wrap nudge + }); + + it('run-log auto-append happens exactly once across a nudged turn', () => { + const start = maxSeq(); + let nudged = false; + + // Result 1: the agent ended the fire with an inert block. + const first = dispatchResultText('digest', taskRouting); + const willRetry = shouldNudgeTaskBlocks(true, first.taskBlocks, nudged); + expect(willRetry).toBe(true); + if (!willRetry) autoAppendTaskLog('digest', start); + nudged = true; + + // Result 2 (post-nudge retry): plain final text → this one becomes the log. + const second = dispatchResultText('Sent the digest to family via send_message.', taskRouting); + const willRetry2 = shouldNudgeTaskBlocks(true, second.taskBlocks, nudged); + expect(willRetry2).toBe(false); + if (!willRetry2) autoAppendTaskLog('Sent the digest to family via send_message.', start); + + const rows = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").all() as { + content: string; + }[]; + expect(rows).toHaveLength(1); + expect(JSON.parse(rows[0].content).text).toBe('Sent the digest to family via send_message.'); + }); +}); + +describe('task-fire run-log auto-append', () => { + it('writes a task_log row from the final text', () => { + const start = maxSeq(); + + autoAppendTaskLog('Checked the\nfeeds — nothing new.', start); + + const rows = getOutboundDb().prepare("SELECT kind, content FROM messages_out WHERE kind = 'task_log'").all() as { + kind: string; + content: string; + }[]; + expect(rows).toHaveLength(1); + expect(JSON.parse(rows[0].content).text).toBe('Checked the feeds — nothing new.'); // whitespace collapsed + }); + + it('strips blocks — logs their inner text marked undelivered, never raw XML', () => { + const start = maxSeq(); + + autoAppendTaskLog('Digest done. 3 new posts today See you tomorrow.', start); + + const rows = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").all() as { + content: string; + }[]; + expect(rows).toHaveLength(1); + const line = JSON.parse(rows[0].content).text as string; + expect(line).not.toContain(' { + const start = maxSeq(); + // The ncl binary writes each CLI call as a cli_request system row. + writeMessageOut({ + id: 'cli-1', + kind: 'system', + content: JSON.stringify({ action: 'cli_request', requestId: 'cli-1', command: 'tasks-append-log', args: { msg: 'done' } }), + }); + expect(hasAppendLogRequestSince(start)).toBe(true); + + autoAppendTaskLog('final text', start); + + const rows = getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all(); + expect(rows).toHaveLength(0); + }); + + it('a positional append-log invocation (dash-joined command) also suppresses', () => { + const start = maxSeq(); + // `ncl tasks append-log "did the thing"` → command 'tasks-append-log-did-the-thing' + writeMessageOut({ + id: 'cli-pos', + kind: 'system', + content: JSON.stringify({ + action: 'cli_request', + requestId: 'cli-pos', + command: 'tasks-append-log-did-the-thing', + args: {}, + }), + }); + + expect(hasAppendLogRequestSince(start)).toBe(true); + + autoAppendTaskLog('final text', start); + expect(getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all()).toHaveLength(0); + }); + + it('a DEFINITIVELY failed append-log (response ok:false) does not suppress', () => { + const start = maxSeq(); + writeMessageOut({ + id: 'cli-fail', + kind: 'system', + content: JSON.stringify({ action: 'cli_request', requestId: 'cli-fail', command: 'tasks-append-log', args: {} }), + }); + // Host response frame lands in inbound messages_in with ok:false. + getInboundDb() + .prepare("INSERT INTO messages_in (id, seq, kind, timestamp, content) VALUES (?, ?, 'system', datetime('now'), ?)") + .run( + 'resp-fail', + 1000, + JSON.stringify({ + requestId: 'cli-fail', + frame: { id: 'cli-fail', ok: false, error: { code: 'invalid-args', message: 'bad series' } }, + }), + ); + + expect(hasAppendLogRequestSince(start)).toBe(false); + + autoAppendTaskLog('final text', start); + expect(getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all()).toHaveLength(1); + }); + + it('an append-log with a still-pending response suppresses (no double-log on the success path)', () => { + const start = maxSeq(); + writeMessageOut({ + id: 'cli-pending', + kind: 'system', + content: JSON.stringify({ + action: 'cli_request', + requestId: 'cli-pending', + command: 'tasks-append-log', + args: {}, + }), + }); + // No response row in inbound yet. + + expect(hasAppendLogRequestSince(start)).toBe(true); + }); + + it('append-log from BEFORE the fire does not suppress', () => { + writeMessageOut({ + id: 'cli-old', + kind: 'system', + content: JSON.stringify({ action: 'cli_request', requestId: 'cli-old', command: 'tasks-append-log', args: {} }), + }); + const start = maxSeq(); // watermark taken after the old call + + autoAppendTaskLog('final text', start); + + const rows = getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all(); + expect(rows).toHaveLength(1); + }); +}); diff --git a/src/cli/resources/tasks.ts b/src/cli/resources/tasks.ts index 2d5f56579a7..4a7676a4419 100644 --- a/src/cli/resources/tasks.ts +++ b/src/cli/resources/tasks.ts @@ -26,6 +26,7 @@ import { import { inboundDbPath, resolveTaskSession, withInboundDb } from '../../session-manager.js'; import { parseZonedToUtc } from '../../timezone.js'; import { registerResource } from '../crud.js'; +import { appendRunLog } from '../../modules/scheduling/run-log.js'; import { formatTasksTable } from '../format-tasks.js'; import type { CallerContext } from '../frame.js'; @@ -284,11 +285,9 @@ function createTask(args: Record, ctx: CallerContext) { 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 — … 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 ""\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` + + `[Task delivery contract:\n` + + `• MESSAGE (only if the task asks you to report/notify): use send_message({ to: "name", … }) with an explicit destination — that tool call is the ONLY thing the user receives. This run has no chat attached: final text and blocks are NOT delivered here.\n` + + `• RUN LOG (automatic): your final text is recorded verbatim in tasks/${id}.md — end the run with a concrete work-log line: what you did and WHY (a no-op run still ends with why nothing was needed; name any files you wrote). Not a greeting, not a copy of the message you sent. For extra mid-run notes use \`ncl tasks append-log --msg "…"\` — if you do, your final text is not auto-logged. Do NOT edit tasks/${id}.md by hand; the log never goes to the user.\n` + `Need context from past runs? Read tasks/${id}.md first.]`; const created = withInbound(session, (db) => { @@ -329,22 +328,12 @@ function appendTaskLog( } } 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 }; + // Group scope is 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. appendRunLog guards the charset. + return { ...appendRunLog(group, series, msg), ok: true }; } /** @@ -659,7 +648,7 @@ registerResource({ 'append-log': { access: 'open', description: - 'Append a one-line run summary to a task run log (tasks/.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.', + 'Append a one-line note to a task run log (tasks/.md).\n\nOptional: a task fire auto-logs its final text, so most runs need no explicit call — use this for mid-run notes (calling it suppresses the final-text auto-log for that fire). The 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.', 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"`, ], diff --git a/src/db/schema.ts b/src/db/schema.ts index 751ca4a8922..53daa983fbc 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -219,7 +219,11 @@ CREATE TABLE IF NOT EXISTS session_routing ( id INTEGER PRIMARY KEY CHECK (id = 1), channel_type TEXT, platform_id TEXT, - thread_id TEXT + thread_id TEXT, + -- 1 = this session is a task-series session (thread system:tasks:*). + -- Stamped by the host so the container never has to sniff the magic + -- thread prefix (which fails open when thread_id is absent/renamed). + is_task INTEGER NOT NULL DEFAULT 0 ); `; diff --git a/src/db/session-db.test.ts b/src/db/session-db.test.ts index 043addb1edf..f1c2ad3fa60 100644 --- a/src/db/session-db.test.ts +++ b/src/db/session-db.test.ts @@ -10,7 +10,13 @@ import fs from 'fs'; import path from 'path'; import { describe, it, expect, afterEach } from 'vitest'; -import { ensureSchema, getInboundSourceSessionId, migrateMessagesInTable, syncProcessingAcks } from './session-db.js'; +import { + ensureSchema, + getInboundSourceSessionId, + migrateMessagesInTable, + syncProcessingAcks, + upsertSessionRouting, +} from './session-db.js'; const TEST_DIR = '/tmp/nanoclaw-session-db-test'; const DB_PATH = path.join(TEST_DIR, 'inbound.db'); @@ -93,6 +99,49 @@ describe('migrateMessagesInTable', () => { }); }); +describe('session_routing is_task stamp', () => { + it('upsert writes is_task, migrating a legacy table without the column', () => { + if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); + fs.mkdirSync(TEST_DIR, { recursive: true }); + + // Legacy session_routing (pre-is_task). + const db = new Database(DB_PATH); + db.exec(`CREATE TABLE session_routing ( + id INTEGER PRIMARY KEY CHECK (id = 1), + channel_type TEXT, platform_id TEXT, thread_id TEXT + )`); + + upsertSessionRouting(db, { + channel_type: null, + platform_id: null, + thread_id: 'system:tasks:daily-abc1', + is_task: 1, + }); + // Idempotent re-run flips it back off (host overwrites on every wake). + upsertSessionRouting(db, { channel_type: 'telegram', platform_id: 't:1', thread_id: null, is_task: 0 }); + + const row = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; + expect(row.is_task).toBe(0); + + upsertSessionRouting(db, { channel_type: null, platform_id: null, thread_id: null, is_task: 1 }); + const row2 = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; + expect(row2.is_task).toBe(1); + db.close(); + }); + + it('fresh inbound schema already carries is_task', () => { + if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); + fs.mkdirSync(TEST_DIR, { recursive: true }); + ensureSchema(DB_PATH, 'inbound'); + const db = new Database(DB_PATH); + const cols = (db.prepare("PRAGMA table_info('session_routing')").all() as Array<{ name: string }>).map( + (c) => c.name, + ); + expect(cols).toContain('is_task'); + db.close(); + }); +}); + describe('syncProcessingAcks — script-skip counter', () => { function freshPair() { if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); diff --git a/src/db/session-db.ts b/src/db/session-db.ts index 3424fce659e..ff9ce796c00 100644 --- a/src/db/session-db.ts +++ b/src/db/session-db.ts @@ -40,17 +40,31 @@ export function openOutboundDbRw(dbPath: string): Database.Database { return db; } +/** + * Lazy, on-open migration for session_routing (mirrors migrateMessagesInTable): + * pre-existing session DBs lack the is_task column. Idempotent; no-op when the + * table itself is missing (older DBs error on upsert exactly as before). + */ +export function migrateSessionRoutingTable(db: Database.Database): void { + const cols = (db.prepare("PRAGMA table_info('session_routing')").all() as Array<{ name: string }>).map((c) => c.name); + if (cols.length > 0 && !cols.includes('is_task')) { + db.prepare('ALTER TABLE session_routing ADD COLUMN is_task INTEGER NOT NULL DEFAULT 0').run(); + } +} + export function upsertSessionRouting( db: Database.Database, - routing: { channel_type: string | null; platform_id: string | null; thread_id: string | null }, + routing: { channel_type: string | null; platform_id: string | null; thread_id: string | null; is_task: 0 | 1 }, ): void { + migrateSessionRoutingTable(db); db.prepare( - `INSERT INTO session_routing (id, channel_type, platform_id, thread_id) - VALUES (1, @channel_type, @platform_id, @thread_id) + `INSERT INTO session_routing (id, channel_type, platform_id, thread_id, is_task) + VALUES (1, @channel_type, @platform_id, @thread_id, @is_task) ON CONFLICT(id) DO UPDATE SET channel_type = excluded.channel_type, platform_id = excluded.platform_id, - thread_id = excluded.thread_id`, + thread_id = excluded.thread_id, + is_task = excluded.is_task`, ).run(routing); } diff --git a/src/delivery.test.ts b/src/delivery.test.ts index 159d9708a87..8fc84d286ad 100644 --- a/src/delivery.test.ts +++ b/src/delivery.test.ts @@ -21,7 +21,7 @@ vi.mock('./container-runner.js', () => ({ vi.mock('./config.js', async () => { const actual = await vi.importActual('./config.js'); - return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-delivery' }; + return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-delivery', GROUPS_DIR: '/tmp/nanoclaw-test-delivery/groups' }; }); const TEST_DIR = '/tmp/nanoclaw-test-delivery'; @@ -35,7 +35,7 @@ import { createMessagingGroupAgent, } from './db/index.js'; import { getDeliveredIds } from './db/session-db.js'; -import { resolveSession, outboundDbPath, openInboundDb } from './session-manager.js'; +import { resolveSession, resolveTaskSession, outboundDbPath, openInboundDb } from './session-manager.js'; import { deliverSessionMessages, setDeliveryAdapter } from './delivery.js'; function now(): string { @@ -341,3 +341,35 @@ describe('deliverSessionMessages — permission check', () => { expect(delivered.has('out-unauth')).toBe(true); }); }); + +describe('deliverSessionMessages — task_log rows (one-door task delivery)', () => { + it('appends to the series run log and never calls the adapter', async () => { + seedAgentAndChannel(); + const { session } = resolveTaskSession('ag-1', 'daily-digest-a1b2'); + + const db = new Database(outboundDbPath('ag-1', session.id)); + db.prepare( + `INSERT INTO messages_out (id, timestamp, kind, content) + VALUES ('log-1', datetime('now'), 'task_log', ?)`, + ).run(JSON.stringify({ text: 'checked feeds; nothing new' })); + db.close(); + + const calls: string[] = []; + setDeliveryAdapter({ + async deliver(_c, _p, _t, _k, content) { + calls.push(content); + return 'pm'; + }, + }); + await deliverSessionMessages(session); + + expect(calls).toHaveLength(0); // a run-log line is not a delivery + const logFile = `${TEST_DIR}/groups/test-agent/tasks/daily-digest-a1b2.md`; + expect(fs.existsSync(logFile)).toBe(true); + const line = fs.readFileSync(logFile, 'utf8').trim(); + expect(line).toMatch(/^\d{4}-\d{2}-\d{2}T.* — checked feeds; nothing new$/); + // Marked delivered — the row is not retried. + const delivered = getDeliveredIds(openInboundDb('ag-1', session.id)); + expect(delivered.has('log-1')).toBe(true); + }); +}); diff --git a/src/delivery.ts b/src/delivery.ts index d9a0d15f270..474c02fb1c6 100644 --- a/src/delivery.ts +++ b/src/delivery.ts @@ -9,7 +9,14 @@ */ import type Database from 'better-sqlite3'; -import { getRunningSessions, getActiveSessions, createPendingQuestion } from './db/sessions.js'; +import { + getRunningSessions, + getActiveSessions, + createPendingQuestion, + isTaskThread, + TASKS_SYSTEM_THREAD_ID, +} from './db/sessions.js'; +import { appendRunLog } from './modules/scheduling/run-log.js'; import { getAgentGroup } from './db/agent-groups.js'; import { getDb, hasTable } from './db/connection.js'; import { getMessagingGroup, getMessagingGroupByPlatform } from './db/messaging-groups.js'; @@ -260,6 +267,24 @@ async function deliverMessage( return; } + // Task-fire run log: the runner mirrors a fire's final text here (one-door + // delivery — final text never reaches a channel; the send_message tool is + // the only delivery path from a task session). Append to the series log, + // never deliver. The caller marks it delivered so it isn't retried. + if (msg.kind === 'task_log') { + if (session.messaging_group_id === null && isTaskThread(session.thread_id) && session.thread_id) { + const series = session.thread_id.slice(`${TASKS_SYSTEM_THREAD_ID}:`.length); + try { + appendRunLog(session.agent_group_id, series, typeof content.text === 'string' ? content.text : ''); + } catch (err) { + log.warn('Failed to append task run log', { id: msg.id, sessionId: session.id, err }); + } + } else { + log.warn('task_log row outside a task session — ignoring', { id: msg.id, sessionId: session.id }); + } + return; + } + // Agent-to-agent — route to target session via the agent-to-agent module. // Guarded by the channel_type check. If the module isn't installed the // `agent_destinations` table won't exist and `routeAgentMessage`'s permission diff --git a/src/host-core.test.ts b/src/host-core.test.ts index 5bee875f4fc..81b3635e919 100644 --- a/src/host-core.test.ts +++ b/src/host-core.test.ts @@ -19,6 +19,7 @@ import { } from './db/index.js'; import { resolveSession, + resolveTaskSession, writeSessionMessage, writeSessionRouting, initSessionFolder, @@ -953,6 +954,32 @@ describe('writeSessionRouting', () => { expect(row!.thread_id).toBeNull(); }); + it('stamps is_task=1 for a task-series session and 0 for chat sessions', () => { + createAgentGroup({ + id: 'ag-1', + name: 'Agent', + folder: 'agent', + agent_provider: null, + created_at: now(), + }); + + const { session: taskSession } = resolveTaskSession('ag-1', 'daily-abc1'); + writeSessionRouting('ag-1', taskSession.id); + + let db = new Database(inboundDbPath('ag-1', taskSession.id)); + const taskRow = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; + db.close(); + expect(taskRow.is_task).toBe(1); + + const { session: chatSession } = resolveSession('ag-1', null, null, 'agent-shared'); + writeSessionRouting('ag-1', chatSession.id); + + db = new Database(inboundDbPath('ag-1', chatSession.id)); + const chatRow = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; + db.close(); + expect(chatRow.is_task).toBe(0); + }); + it('includes thread_id from per-thread session', () => { createAgentGroup({ id: 'ag-1', diff --git a/src/modules/scheduling/recurrence.test.ts b/src/modules/scheduling/recurrence.test.ts index c470e5a2b73..a937dc05adb 100644 --- a/src/modules/scheduling/recurrence.test.ts +++ b/src/modules/scheduling/recurrence.test.ts @@ -19,9 +19,15 @@ import type { Session } from '../../types.js'; // 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(); - return { ...actual, TIMEZONE: 'Asia/Tokyo' }; + return { ...actual, TIMEZONE: 'Asia/Tokyo', GROUPS_DIR: '/tmp/nanoclaw-recurrence-test/groups' }; }); +// The auto-pause note goes through the shared appendRunLog helper, which +// resolves the group folder from the central DB — mock it to a fixed folder. +vi.mock('../../db/agent-groups.js', () => ({ + getAgentGroup: (id: string) => (id === 'ag-test' ? { id, folder: 'g-test' } : undefined), +})); + const TEST_DIR = '/tmp/nanoclaw-recurrence-test'; const DB_PATH = path.join(TEST_DIR, 'inbound.db'); @@ -190,4 +196,18 @@ describe('handleRecurrence — script-failure backoff (streak derived from faile }; expect(original.recurrence).toBeNull(); // not re-cloned next sweep }); + + it('writes the auto-pause note into the series run log via the shared appendRunLog', async () => { + const db = freshDb(); + seedFailedStreak(db, 8); + await handleRecurrence(db, fakeSession()); + + // Same file + format appendRunLog owns: groups//tasks/.md + const logFile = path.join(TEST_DIR, 'groups', 'g-test', 'tasks', 'task-s-0.md'); + expect(fs.existsSync(logFile)).toBe(true); + const content = fs.readFileSync(logFile, 'utf8'); + expect(content).toContain('auto-paused after 8 consecutive script failures'); + expect(content).toContain('ncl tasks resume task-s-0'); + expect(content).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z — /m); // appendRunLog's stamp + }); }); diff --git a/src/modules/scheduling/recurrence.ts b/src/modules/scheduling/recurrence.ts index 0ebecd5fe94..99e68945448 100644 --- a/src/modules/scheduling/recurrence.ts +++ b/src/modules/scheduling/recurrence.ts @@ -11,17 +11,14 @@ * direct dynamic import. When scheduling moves to the modules branch in * PR #8, the install skill re-fills the marker on install. */ -import fs from 'fs'; -import path from 'path'; - 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 { TIMEZONE } from '../../config.js'; import { log } from '../../log.js'; import type { Session } from '../../types.js'; import { clearRecurrence, getCompletedRecurring, insertRecurrence, trailingFailedRuns } from './db.js'; +import { appendRunLog } from './run-log.js'; // Consecutive pre-task-script failures (the series' trailing FAILED runs — // derived from occurrence rows, no stored counter) throttle a broken monitor @@ -38,15 +35,16 @@ export function scriptBackoffMinutes(fails: number): number { } /** 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). */ + * append-log when a script-gated series is auto-paused. Uses the shared + * appendRunLog helper (one writer format); appendRunLog throws on a bad + * series charset or a missing agent group, and the sweep must not crash + * over a log line, so failures are logged and swallowed. */ 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`); + try { + appendRunLog(agentGroupId, seriesId, note); + } catch (err) { + log.warn('Could not append host task note to run log', { agentGroupId, seriesId, err }); + } } export async function handleRecurrence(inDb: Database.Database, session: Session): Promise { diff --git a/src/modules/scheduling/run-log.ts b/src/modules/scheduling/run-log.ts new file mode 100644 index 00000000000..3684a801b66 --- /dev/null +++ b/src/modules/scheduling/run-log.ts @@ -0,0 +1,32 @@ +/** + * Task-series run log — one host-timestamped line per event, at + * `//tasks/.md`. + * + * Two writers, one format: + * - `ncl tasks append-log` (agent's explicit mid-run/work-log entry) + * - the `task_log` outbound row a task fire's final text produces + * (container/agent-runner poll-loop auto-append; delivery.ts routes it here) + */ +import fs from 'fs'; + +import { GROUPS_DIR } from '../../config.js'; +import { getAgentGroup } from '../../db/agent-groups.js'; + +export function appendRunLog( + agentGroupId: string, + series: string, + msg: string, +): { series: string; timestamp: string; path: string } { + // Charset guard is the security boundary: blocks path traversal and keeps + // the id safe as a filename. Callers resolve group scope before this. + if (!/^[a-z0-9-]+$/.test(series)) throw new Error(`invalid task id: ${series}`); + const ag = getAgentGroup(agentGroupId); + if (!ag) throw new Error(`agent group not found: ${agentGroupId}`); + + 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 }; +} diff --git a/src/session-manager.ts b/src/session-manager.ts index c8ca725a7c6..62dff1b4128 100644 --- a/src/session-manager.ts +++ b/src/session-manager.ts @@ -26,6 +26,7 @@ import { findSessionByAgentGroup, findSessionForAgent, getSession, + isTaskThread, taskThreadId, updateSession, } from './db/sessions.js'; @@ -193,6 +194,10 @@ export function writeSessionRouting(agentGroupId: string, sessionId: string): vo channel_type: channelType, platform_id: platformId, thread_id: session.thread_id, + // Stamp task-ness explicitly so the container can gate one-door + // delivery on a host-asserted flag instead of sniffing the thread + // prefix (fail-closed at the source of truth). + is_task: isTaskThread(session.thread_id) ? 1 : 0, }); } finally { db.close(); From 5512731512a0bf77c49138448f051aaeb2972b20 Mon Sep 17 00:00:00 2001 From: Omri Maya Date: Fri, 10 Jul 2026 10:36:52 +0300 Subject: [PATCH 2/7] fix(tasks): simplify one-door delivery contract --- CHANGELOG.md | 2 +- .../src/compact-instructions.test.ts | 21 ++ .../agent-runner/src/compact-instructions.ts | 55 ++-- container/agent-runner/src/db/messages-out.ts | 60 ---- .../agent-runner/src/db/session-routing.ts | 38 +-- .../agent-runner/src/destinations.test.ts | 13 + container/agent-runner/src/destinations.ts | 38 ++- container/agent-runner/src/formatter.test.ts | 34 ++- container/agent-runner/src/formatter.ts | 29 +- container/agent-runner/src/index.ts | 7 +- .../src/mcp-tools/cli.instructions.md | 6 +- .../src/mcp-tools/core.instructions.md | 18 +- container/agent-runner/src/mcp-tools/core.ts | 62 +--- .../src/mcp-tools/scheduling.instructions.md | 2 +- container/agent-runner/src/poll-loop.ts | 114 ++++--- .../agent-runner/src/task-delivery.test.ts | 284 +++++++----------- src/cli/resources/destinations.ts | 2 +- src/cli/resources/tasks.test.ts | 4 +- src/cli/resources/tasks.ts | 21 +- src/container-runner.ts | 2 +- src/db/schema.ts | 12 +- src/db/session-db.test.ts | 51 +--- src/db/session-db.ts | 22 +- src/delivery.ts | 2 +- src/host-core.test.ts | 27 -- src/modules/agent-to-agent/create-agent.ts | 2 +- src/modules/scheduling/run-log.ts | 2 +- src/session-manager.ts | 11 +- 28 files changed, 407 insertions(+), 534 deletions(-) create mode 100644 container/agent-runner/src/compact-instructions.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 71fd4bfec99..1c68c65f5bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ All notable changes to NanoClaw will be documented in this file. ## [Unreleased] -- [BREAKING] **One-door delivery in task sessions.** A task fire delivers ONLY via the `send_message` tool, and `to` is required there (no default destination). Final-text `` blocks are inert in task sessions, and the fire's final text is auto-appended to the series run log (`tasks/.md`) unless the agent already ran `ncl tasks append-log` that fire — a run is logged exactly once, and the double-send/zero-send ambiguity of having two delivery paths is gone (the previous echo-suppression heuristics are removed). Chat sessions are unchanged. **Migration:** rebuild the agent container image (`./container/build.sh`) and restart so containers pick up the new runner; task prompts that relied on `` blocks must switch to `send_message(to=...)` — existing tasks created with the standard scaffold already instruct this. +- [BREAKING] **Explicit destinations and one-door task delivery.** Every `send_message` and `send_file` call now requires a named `to` destination; there is no single-destination or reply-in-place shortcut. In task sessions, only explicitly addressed tool calls deliver, while final output becomes the automatic run summary and `ncl tasks append-log` adds optional progress notes. Existing task DBs need no schema migration; legacy generated task instructions are normalized when read. **Migration:** rebuild the agent image, restart NanoClaw, and update custom instructions that omit `to`. - **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n−1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. - [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. diff --git a/container/agent-runner/src/compact-instructions.test.ts b/container/agent-runner/src/compact-instructions.test.ts new file mode 100644 index 00000000000..fa3537b9022 --- /dev/null +++ b/container/agent-runner/src/compact-instructions.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from 'bun:test'; + +import { buildCompactInstructions } from './compact-instructions.js'; + +describe('compaction delivery reminder', () => { + it('preserves final-output addressing in chat sessions', () => { + const instructions = buildCompactInstructions(['family'], null); + + expect(instructions).toContain(''); + expect(instructions).toContain('`family`'); + }); + + it('preserves explicit-tool delivery in task sessions without teaching final-output blocks', () => { + const instructions = buildCompactInstructions(['family'], 'daily-digest-a1b2'); + + expect(instructions).toContain('send_message'); + expect(instructions).toContain('explicit to destination'); + expect(instructions).toContain('tasks/daily-digest-a1b2.md'); + expect(instructions).not.toContain(''); + }); +}); diff --git a/container/agent-runner/src/compact-instructions.ts b/container/agent-runner/src/compact-instructions.ts index 1cbea0dd583..38bcf37376a 100644 --- a/container/agent-runner/src/compact-instructions.ts +++ b/container/agent-runner/src/compact-instructions.ts @@ -10,25 +10,42 @@ * "command": "bun /app/src/compact-instructions.ts" */ import { getAllDestinations } from './destinations.js'; +import { getTaskSeriesId } from './db/session-routing.js'; -const destinations = getAllDestinations(); -const names = destinations.map((d) => d.name); +export function buildCompactInstructions(names: string[], taskId: string | null): string { + const deliveryReminder = taskId + ? [ + ' "This is an isolated task run. Send a user-visible message only with send_message and an explicit to destination.', + ` Final output is not delivered; it becomes the automatic summary in tasks/${taskId}.md.`, + ` Available destinations: ${formatDestinationNames(names)}."`, + ] + : [ + ' "You MUST wrap all responses in ... blocks.', + ` Available destinations: ${formatDestinationNames(names)}."`, + ]; -const instructions = [ - 'Preserve the following in the compaction summary:', - '', - '1. For recent messages, keep the full XML structure including all attributes:', - ' - for chat messages', - ' - for scheduled tasks', - ' - for webhooks', - ' The message content can be summarized if long, but the XML tags and attributes must remain.', - '', - '2. Preserve the chronological message/reply sequence of recent exchanges.', - ' The agent needs to see: who said what, in what order, and from which destination.', - '', - '3. At the END of the compaction summary, include this verbatim reminder:', - ' "You MUST wrap all responses in ... blocks.', - ` Available destinations: ${names.length > 0 ? names.map((n) => `\`${n}\``).join(', ') : '(none)'}."`, -]; + return [ + 'Preserve the following in the compaction summary:', + '', + '1. For recent messages, keep the full XML structure including all attributes:', + ' - for chat messages', + ' - for scheduled tasks', + ' - for webhooks', + ' The message content can be summarized if long, but the XML tags and attributes must remain.', + '', + '2. Preserve the chronological message/reply sequence of recent exchanges.', + ' The agent needs to see: who said what, in what order, and from which destination.', + '', + '3. At the END of the compaction summary, include this verbatim reminder:', + ...deliveryReminder, + ].join('\n'); +} -console.log(instructions.join('\n')); +function formatDestinationNames(names: string[]): string { + return names.length > 0 ? names.map((name) => `\`${name}\``).join(', ') : '(none)'; +} + +if (import.meta.main) { + const names = getAllDestinations().map((destination) => destination.name); + console.log(buildCompactInstructions(names, getTaskSeriesId())); +} diff --git a/container/agent-runner/src/db/messages-out.ts b/container/agent-runner/src/db/messages-out.ts index 9568e8b3c2f..85b04671c68 100644 --- a/container/agent-runner/src/db/messages-out.ts +++ b/container/agent-runner/src/db/messages-out.ts @@ -141,63 +141,3 @@ export function getUndeliveredMessages(): MessageOutRow[] { ) .all() as MessageOutRow[]; } - -/** Highest seq across both session DBs — marks "now" for since-queries. */ -export function maxSeq(): number { - const maxOut = (getOutboundDb().prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_out').get() as { m: number }) - .m; - const maxIn = (getInboundDb().prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_in').get() as { m: number }).m; - return Math.max(maxOut, maxIn); -} - -/** - * True if the agent invoked `ncl tasks append-log` after `sinceSeq` (the ncl - * binary writes each CLI call as a system cli_request row in messages_out). - * Guards the task-fire auto-log: an explicit append-log this fire suppresses - * the runner's final-text auto-append, so a run is logged exactly once. - */ -export function hasAppendLogRequestSince(sinceSeq: number): boolean { - // LIKE, not equality: a positional invocation (`ncl tasks append-log "..."`) - // arrives dash-joined as 'tasks-append-log-' — the host dispatcher - // resolves the tail as the positional id. - const requests = getOutboundDb() - .prepare( - `SELECT content FROM messages_out - WHERE seq > $since AND kind = 'system' - AND json_extract(content, '$.action') = 'cli_request' - AND json_extract(content, '$.command') LIKE 'tasks-append-log%'`, - ) - .all({ $since: sinceSeq }) as Array<{ content: string }>; - if (requests.length === 0) return false; - - // Only a request that SUCCEEDED counts as "this fire was logged". Correlate - // each request with its host response in inbound messages_in (content carries - // the requestId + response frame). A request with no readable response yet - // still suppresses — the common success case is the response landing after - // the turn ends, and double-logging is worse than a rare missed line. Only a - // DEFINITIVE failure (frame ok:false) does not suppress. - const inbound = getInboundDb(); - for (const { content } of requests) { - let requestId: string | null = null; - try { - requestId = (JSON.parse(content) as { requestId?: string }).requestId ?? null; - } catch { - // Malformed request row — can't correlate; suppress conservatively. - return true; - } - if (!requestId) return true; - - const resp = inbound - .prepare('SELECT content FROM messages_in WHERE content LIKE $pat ORDER BY seq DESC LIMIT 1') - .get({ $pat: `%"requestId":"${requestId}"%` }) as { content: string } | undefined; - if (!resp) return true; // pending response → treat as logged - try { - const frame = (JSON.parse(resp.content) as { frame?: { ok?: boolean } }).frame; - if (frame?.ok !== false) return true; // ok:true (or unreadable frame) → logged - } catch { - return true; // unparseable response — suppress conservatively - } - // frame.ok === false → this request definitively failed; check the next one. - } - return false; -} diff --git a/container/agent-runner/src/db/session-routing.ts b/container/agent-runner/src/db/session-routing.ts index d5d24633191..ab52b6a757d 100644 --- a/container/agent-runner/src/db/session-routing.ts +++ b/container/agent-runner/src/db/session-routing.ts @@ -1,12 +1,9 @@ /** - * Default reply routing for this session — written by the host on every + * Current chat/thread routing for this session — written by the host on every * container wake (see src/session-manager.ts `writeSessionRouting`). * - * Read by the MCP tools as the default destination for outbound messages - * when the agent doesn't specify an explicit `to`. This is what makes - * "agent replies in the thread it's currently in" work: the router strips - * or preserves thread_id based on the adapter's thread support, and we - * just read the fixed routing the host committed for this session. + * Read by MCP tools to preserve the current thread when an explicitly named + * destination resolves to the chat this session is bound to. */ import { getInboundDb } from './connection.js'; @@ -14,28 +11,25 @@ export interface SessionRouting { channel_type: string | null; platform_id: string | null; thread_id: string | null; - /** 1 = host stamped this session as a task-series session. */ - is_task: 0 | 1; } export function getSessionRouting(): SessionRouting { const db = getInboundDb(); try { - const row = db - .prepare('SELECT channel_type, platform_id, thread_id, is_task FROM session_routing WHERE id = 1') - .get() as SessionRouting | undefined; + const row = db.prepare('SELECT channel_type, platform_id, thread_id FROM session_routing WHERE id = 1').get() as + | SessionRouting + | undefined; if (row) return row; } catch { - // is_task column may not exist yet (host predates the stamp) — retry - // without it and derive task-ness from the thread prefix. - try { - const row = db - .prepare('SELECT channel_type, platform_id, thread_id FROM session_routing WHERE id = 1') - .get() as Omit | undefined; - if (row) return { ...row, is_task: row.thread_id?.startsWith('system:tasks') ? 1 : 0 }; - } catch { - // Table may not exist on an older session DB — fall through to defaults - } + // Table may not exist on an older session DB — fall through to defaults. } - return { channel_type: null, platform_id: null, thread_id: null, is_task: 0 }; + return { channel_type: null, platform_id: null, thread_id: null }; +} + +const TASK_THREAD_PREFIX = 'system:tasks:'; + +/** The task id encoded in this isolated task session's canonical thread id. */ +export function getTaskSeriesId(): string | null { + const threadId = getSessionRouting().thread_id; + return threadId?.startsWith(TASK_THREAD_PREFIX) ? threadId.slice(TASK_THREAD_PREFIX.length) : null; } diff --git a/container/agent-runner/src/destinations.test.ts b/container/agent-runner/src/destinations.test.ts index 61e5f4bb39c..aa016a09699 100644 --- a/container/agent-runner/src/destinations.test.ts +++ b/container/agent-runner/src/destinations.test.ts @@ -60,4 +60,17 @@ describe('buildSystemPromptAddendum — multi-destination routing guidance', () expect(prompt).toContain('default to addressing the destination it came `from`'); expect(prompt).toContain('`casa`'); }); + + it('gives task sessions only explicit-tool delivery instructions', () => { + seedDestination('casa', 'Casa', 'whatsapp', 'group-1@g.us'); + + const prompt = buildSystemPromptAddendum('Casa', { kind: 'task', taskId: 'daily-briefing-a25c' }); + + expect(prompt).toContain('isolated task run'); + expect(prompt).toContain('send_message({ to: "name"'); + expect(prompt).toContain('tasks/daily-briefing-a25c.md'); + expect(prompt).toContain('Only notify someone when the task asks'); + expect(prompt).not.toContain('…` block; include several blocks in one response to address several destinations. `…` marks thinking you don\'t want sent.', ); @@ -122,7 +130,11 @@ function buildDestinationsSection(): string { ); lines.push(''); lines.push( - 'The `send_message` MCP tool is the same delivery, available mid-turn — handy for a quick acknowledgment ("on it") before a slow tool call. Each `send_message` call and each final-response `` block lands as its own message in the conversation, so they read as a sequence rather than as one combined reply.', + 'The `send_message` MCP tool is the same delivery, available mid-turn — handy for a quick acknowledgment ("on it") before a slow tool call. Always pass its explicit `to` destination. Each `send_message` call and each final-response `` block lands as its own message in the conversation, so they read as a sequence rather than as one combined reply.', + ); + lines.push(''); + lines.push( + 'For a short turn, do not narrate. For longer work, send one acknowledgment and then updates only at meaningful milestones, especially before slow operations. Never narrate micro-steps; finish with the outcome, not a play-by-play.', ); return lines.join('\n'); } diff --git a/container/agent-runner/src/formatter.test.ts b/container/agent-runner/src/formatter.test.ts index 9121366f8a2..a18ad6d2adf 100644 --- a/container/agent-runner/src/formatter.test.ts +++ b/container/agent-runner/src/formatter.test.ts @@ -13,7 +13,7 @@ import { describe, it, expect, beforeEach, afterEach } from 'bun:test'; import { initTestSessionDb, closeSessionDb, getInboundDb } from './db/connection.js'; import { getPendingMessages } from './db/messages-in.js'; -import { formatMessages, stripInternalTags } from './formatter.js'; +import { formatMessages, stripInternalTags, stripLegacyTaskContract } from './formatter.js'; import { TIMEZONE } from './timezone.js'; beforeEach(() => { @@ -62,6 +62,38 @@ describe('context timezone header', () => { }); }); +describe('task prompt compatibility', () => { + it('strips the generated #2981 delivery suffix without mutating stored data', () => { + const prompt = + 'Send the daily digest\n\n' + + '[A task serves the user two separate ways — legacy generated delivery instructions]'; + + expect(stripLegacyTaskContract(prompt)).toBe('Send the daily digest'); + }); + + it('strips the generated #2988 delivery suffix', () => { + const prompt = 'Check the feeds\n\n[Task delivery contract:\nlegacy generated instructions]'; + + expect(stripLegacyTaskContract(prompt)).toBe('Check the feeds'); + }); + + it('leaves ordinary user prompts unchanged', () => { + const prompt = 'Explain [Task delivery contract:] as plain text'; + + expect(stripLegacyTaskContract(prompt)).toBe(prompt); + }); + + it('does not expose a legacy delivery contract in a formatted task run', () => { + insertMessage('task-1', 'task', { + prompt: 'Check the feeds\n\n[Task delivery contract:\nlegacy generated instructions]', + }); + + const result = formatMessages(getPendingMessages()); + expect(result).toContain('Instructions:\nCheck the feeds'); + expect(result).not.toContain('legacy generated instructions'); + }); +}); + describe('multi-message chat batches', () => { // Regression guard for #2555: an outer `` envelope around // multiple chat messages caused the Claude Agent SDK to emit a synthetic diff --git a/container/agent-runner/src/formatter.ts b/container/agent-runner/src/formatter.ts index a8375f5144d..232ab9108ad 100644 --- a/container/agent-runner/src/formatter.ts +++ b/container/agent-runner/src/formatter.ts @@ -98,10 +98,10 @@ export interface RoutingContext { channelType: string | null; threadId: string | null; inReplyTo: string | null; - /** Batch is a task fire. One-door delivery: only the send_message tool + /** Batch is a task run. One-door delivery: only an explicitly addressed tool * delivers from a task session; final-text `` blocks are inert * and the final text auto-appends to the series run log. */ - taskFire: boolean; + taskRun: boolean; } /** @@ -115,7 +115,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'), + taskRun: messages.length > 0 && messages.every((m) => m.kind === 'task'), }; } @@ -208,10 +208,31 @@ function formatTaskMessage(msg: MessageInRow): string { if (content.scriptOutput) { parts.push('Script output:', JSON.stringify(content.scriptOutput, null, 2), ''); } - parts.push('Instructions:', content.prompt || ''); + parts.push('Instructions:', stripLegacyTaskContract(content.prompt || '')); return `${parts.join('\n')}`; } +const LEGACY_TASK_CONTRACT_MARKERS = [ + '\n\n[A task serves the user two separate ways —', + '\n\n[Task delivery contract:', +]; + +/** + * PR #2981 persisted its generated delivery contract inside each task prompt. + * New sessions receive the contract from their runtime system prompt instead. + * Strip only a known generated suffix, at read time, so existing task rows stay + * compatible without a session-DB migration or contradictory model guidance. + */ +export function stripLegacyTaskContract(prompt: string): string { + if (!prompt.trimEnd().endsWith(']')) return prompt; + + let contractStart = -1; + for (const marker of LEGACY_TASK_CONTRACT_MARKERS) { + contractStart = Math.max(contractStart, prompt.lastIndexOf(marker)); + } + return contractStart >= 0 ? prompt.slice(0, contractStart).trimEnd() : prompt; +} + function formatWebhookMessage(msg: MessageInRow): string { const content = parseContent(msg.content); const source = content.source || 'unknown'; diff --git a/container/agent-runner/src/index.ts b/container/agent-runner/src/index.ts index d688cf5e1e1..adc8e392dad 100644 --- a/container/agent-runner/src/index.ts +++ b/container/agent-runner/src/index.ts @@ -27,6 +27,7 @@ import { fileURLToPath } from 'url'; import { loadConfig } from './config.js'; import { buildSystemPromptAddendum } from './destinations.js'; +import { getTaskSeriesId } from './db/session-routing.js'; import { ensureMemoryScaffold } from './memory-scaffold.js'; // Providers barrel — each enabled provider self-registers on import. // Provider skills append imports to providers/index.ts. @@ -52,7 +53,11 @@ async function main(): Promise { // /workspace/agent/CLAUDE.md — the composed entry imports the shared // base (/app/CLAUDE.md) and each enabled module's fragment. Per-group // memory lives in /workspace/agent/CLAUDE.local.md (auto-loaded). - const instructions = buildSystemPromptAddendum(config.assistantName || undefined); + const taskId = getTaskSeriesId(); + const instructions = buildSystemPromptAddendum( + config.assistantName || undefined, + taskId ? { kind: 'task', taskId } : { kind: 'chat' }, + ); // Discover additional directories mounted at /workspace/extra/* const additionalDirectories: string[] = []; diff --git a/container/agent-runner/src/mcp-tools/cli.instructions.md b/container/agent-runner/src/mcp-tools/cli.instructions.md index 40b98671215..db27a38e93f 100644 --- a/container/agent-runner/src/mcp-tools/cli.instructions.md +++ b/container/agent-runner/src/mcp-tools/cli.instructions.md @@ -67,9 +67,9 @@ 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" +# Add an optional progress note during a task run. The final response is logged automatically. +# This is a LOG ENTRY, not a message: it sends nothing to anyone. Inside a task run --id is auto-derived. +ncl tasks append-log --msg "one feed returned 403; continuing with the remaining feeds" # Write commands (approval required) ncl groups restart diff --git a/container/agent-runner/src/mcp-tools/core.instructions.md b/container/agent-runner/src/mcp-tools/core.instructions.md index 85ad24bbd83..43f36fd4c26 100644 --- a/container/agent-runner/src/mcp-tools/core.instructions.md +++ b/container/agent-runner/src/mcp-tools/core.instructions.md @@ -1,22 +1,10 @@ -## Sending messages +## Outbound tools -**Every response** must be wrapped in `...` blocks — even if you only have one destination. Bare text outside of `` blocks is scratchpad (logged but never sent). See the `## Sending messages` section in your runtime system prompt for the current destination list and names. - -### Mid-turn updates (`send_message`) - -Use the `mcp__nanoclaw__send_message` tool to send a message while you're still working (before your final output). If you have one destination, `to` is optional; with multiple, specify it. Pace your updates to the length of the work: - -- **Short turn (≤2 quick tool calls):** Don't narrate. Output any response. -- **Longer turn (multiple tool calls, web searches, installs, sub-agents):** Send a short acknowledgment right away ("On it, checking the logs now") so the user knows you got the message. -- **Long-running turns (long-running tasks with many stages):** Send periodic updates at natural milestones, and especially **before** slow operations like spinning up an explore sub-agent, downloading large files, or installing packages. - -**Never narrate micro-steps.** "I'm going to read the file now… okay, I'm reading it… now I'm parsing it…" is noise. Updates should mark meaningful transitions, not every tool call. - -**Outcomes, not play-by-play.** When the turn is done, the final message should be about the result, not a transcript of what you did. +The runtime system prompt lists your destinations and explains how final output is handled in this session. Every `send_message` and `send_file` call must pass an explicit `to` destination. ### Sending files (`send_file`) -Use `mcp__nanoclaw__send_file({ path, text?, filename?, to? })` to deliver a file from your workspace. `path` is absolute or relative to `/workspace/agent/`; `filename` overrides the display name shown in chat (defaults to the file's basename); `text` is an optional accompanying message. Use this for artifacts you produce (charts, PDFs, generated images, reports) rather than dumping contents into chat. +Use `mcp__nanoclaw__send_file({ to, path, text?, filename? })` to deliver a file from your workspace. `path` is absolute or relative to `/workspace/agent/`; `filename` overrides the display name shown in chat (defaults to the file's basename); `text` is an optional accompanying message. Use this for artifacts you produce (charts, PDFs, generated images, reports) rather than dumping contents into chat. ### Reacting to messages (`add_reaction`) diff --git a/container/agent-runner/src/mcp-tools/core.ts b/container/agent-runner/src/mcp-tools/core.ts index 7d50b86794e..61bd5579c45 100644 --- a/container/agent-runner/src/mcp-tools/core.ts +++ b/container/agent-runner/src/mcp-tools/core.ts @@ -41,51 +41,14 @@ function destinationList(): string { /** * Resolve a destination name to routing fields. * - * If `to` is omitted, use the session's default reply routing (channel + - * thread the conversation is in) — the agent replies in place. - * - * If `to` is specified, look up the named destination. If it resolves to + * Look up the explicitly named destination. If it resolves to * the same channel the session is bound to, the session's thread_id is * preserved so replies land in the correct thread. Otherwise thread_id * is null (a cross-destination send starts a new conversation). */ function resolveRouting( - to: string | undefined, + to: string, ): { channel_type: string; platform_id: string; thread_id: string | null; resolvedName: string } | { error: string } { - if (!to) { - const session = getSessionRouting(); - // Task sessions have no origin chat — "reply in place" is meaningless and - // the single-destination shortcut below would silently pick an arbitrary - // target. One-door rule: a task fire must name its destination. - // Driven by the host-stamped is_task flag (session_routing); the thread - // prefix check is only a fallback for hosts that predate the stamp. - // (Prefix mirrors TASKS_SYSTEM_THREAD_ID on the host, src/db/sessions.ts — - // the two runtimes share no modules.) - if (session.is_task === 1 || session.thread_id?.startsWith('system:tasks')) { - return { - error: `This is a task session — pass to= explicitly. Options: ${destinationList()}`, - }; - } - // Default: reply to whatever thread/channel this session is bound to. - if (session.channel_type && session.platform_id) { - return { - channel_type: session.channel_type, - platform_id: session.platform_id, - thread_id: session.thread_id, - resolvedName: '(current conversation)', - }; - } - // No session routing (e.g., agent-shared or internal-only agent) — - // fall back to the legacy single-destination shortcut. - const all = getAllDestinations(); - if (all.length === 0) return { error: 'No destinations configured.' }; - if (all.length > 1) { - return { - error: `You have multiple destinations — specify "to". Options: ${all.map((d) => d.name).join(', ')}`, - }; - } - to = all[0].name; - } const dest = findByName(to); if (!dest) return { error: `Unknown destination "${to}". Known: ${destinationList()}` }; if (dest.type === 'channel') { @@ -107,25 +70,26 @@ function resolveRouting( export const sendMessage: McpToolDefinition = { tool: { name: 'send_message', - description: - 'Send a message to a named destination. If you have only one destination, you can omit `to` — except in task sessions, where `to` is always required (this tool is the ONLY delivery path there; final text is never sent).', + description: 'Send a message to a named destination.', inputSchema: { type: 'object' as const, properties: { to: { type: 'string', - description: 'Destination name (e.g., "family", "worker-1"). Optional if you have only one destination.', + description: 'Destination name (e.g., "family", "worker-1").', }, text: { type: 'string', description: 'Message content' }, }, - required: ['text'], + required: ['to', 'text'], }, }, async handler(args) { + const to = args.to as string; const text = args.text as string; + if (!to) return err(`to is required. Options: ${destinationList()}`); if (!text) return err('text is required'); - const routing = resolveRouting(args.to as string | undefined); + const routing = resolveRouting(to); if ('error' in routing) return err(routing.error); const id = generateId(); @@ -147,23 +111,25 @@ export const sendMessage: McpToolDefinition = { export const sendFile: McpToolDefinition = { tool: { name: 'send_file', - description: 'Send a file to a named destination. If you have only one destination, you can omit `to`.', + description: 'Send a file to a named destination.', inputSchema: { type: 'object' as const, properties: { - to: { type: 'string', description: 'Destination name. Optional if you have only one destination.' }, + to: { type: 'string', description: 'Destination name.' }, path: { type: 'string', description: 'File path (relative to /workspace/agent/ or absolute)' }, text: { type: 'string', description: 'Optional accompanying message' }, filename: { type: 'string', description: 'Display name (default: basename of path)' }, }, - required: ['path'], + required: ['to', 'path'], }, }, async handler(args) { + const to = args.to as string; const filePath = args.path as string; + if (!to) return err(`to is required. Options: ${destinationList()}`); if (!filePath) return err('path is required'); - const routing = resolveRouting(args.to as string | undefined); + const routing = resolveRouting(to); if ('error' in routing) return err(routing.error); const resolvedPath = path.isAbsolute(filePath) ? filePath : path.resolve('/workspace/agent', filePath); diff --git a/container/agent-runner/src/mcp-tools/scheduling.instructions.md b/container/agent-runner/src/mcp-tools/scheduling.instructions.md index fcb70a72e96..033c4da9d6b 100644 --- a/container/agent-runner/src/mcp-tools/scheduling.instructions.md +++ b/container/agent-runner/src/mcp-tools/scheduling.instructions.md @@ -1,6 +1,6 @@ ## Task scheduling (`ncl tasks`) -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. When it fires, the ONLY way to message anyone is `send_message({ to: "name", ... })` with an explicit destination — final text and `` blocks are not delivered from task sessions; the fire's final text is recorded in the task's run log instead. +Use `ncl tasks` for one-shot and recurring tasks. Each task runs in its own isolated session. Its runtime prompt supplies the task-only delivery and run-log contract. Pass `--name ""` on create to get a readable task id (e.g. `--name "sales briefing"` → `sales-briefing-a25c`); without it ids are `t-`. diff --git a/container/agent-runner/src/poll-loop.ts b/container/agent-runner/src/poll-loop.ts index e0acbce988b..7e304938f2f 100644 --- a/container/agent-runner/src/poll-loop.ts +++ b/container/agent-runner/src/poll-loop.ts @@ -1,6 +1,12 @@ import { findByName, getAllDestinations, type DestinationEntry } from './destinations.js'; -import { getPendingMessages, markProcessing, markCompleted, markScriptSkipped, type MessageInRow } from './db/messages-in.js'; -import { hasAppendLogRequestSince, maxSeq, writeMessageOut } from './db/messages-out.js'; +import { + getPendingMessages, + markProcessing, + markCompleted, + markScriptSkipped, + type MessageInRow, +} from './db/messages-in.js'; +import { writeMessageOut } from './db/messages-out.js'; import { getInboundDb, touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js'; import { clearContinuation, @@ -340,7 +346,7 @@ export async function processQuery( let queryContinuation: string | undefined; let done = false; let unwrappedNudged = false; - // Once-per-turn guard for the task-fire " block was not delivered" + // Once-per-turn guard for the task-run " block was not delivered" // nudge — mirrors unwrappedNudged for chat turns. let taskBlockNudged = false; // Prompt queue for the exchange hook — each result event consumes the @@ -468,10 +474,6 @@ export async function processQuery( })(); }, ACTIVE_POLL_INTERVAL_MS); - // Seq watermark before the agent runs — anything after this is "this fire" - // for the append-log exactly-once guard. - const turnStartSeq = maxSeq(); - try { for await (const event of query.events) { handleEvent(event, routing); @@ -496,14 +498,14 @@ export async function processQuery( markCompleted(initialBatchIds); if (event.text) { const { sent, hasUnwrapped, taskBlocks } = dispatchResultText(event.text, routing); - const willRetryTaskBlocks = shouldNudgeTaskBlocks(routing.taskFire === true, taskBlocks, taskBlockNudged); + const willRetryTaskBlocks = shouldNudgeTaskBlocks(routing.taskRun, taskBlocks, taskBlockNudged); // One-door task delivery: the final text becomes the run log entry - // (guarded — an explicit append-log this fire wins). Errors included: - // a failed fire's text belongs in the run log, not in a chat. - // When we nudge on inert blocks, DEFER the append to the - // retry's result so the fire logs exactly once. - if (routing.taskFire && !willRetryTaskBlocks) autoAppendTaskLog(event.text, turnStartSeq); - if (sent === 0 && event.isError === true && !routing.taskFire) { + // while explicit append-log calls remain optional additive notes. + // Errors included: a failed run's text belongs in its log, not chat. + // A corrective retry handles delivery only; its result is not a + // second run summary. + if (routing.taskRun && !taskBlockNudged) autoAppendTaskLog(event.text); + if (sent === 0 && event.isError === true && !routing.taskRun) { // Non-retryable error turn (e.g. a 403 billing_error) with no // envelope: deliver the notice instead of dropping it as // scratchpad, and skip the re-wrap nudge — it would just re-hammer @@ -522,7 +524,7 @@ export async function processQuery( prompt: archivePrompts[0] ?? initialPrompt, result: event.text, continuation: queryContinuation ?? initialContinuation, - status: hasUnwrapped ? 'undelivered' : 'completed', + status: hasUnwrapped || willRetryTaskBlocks ? 'undelivered' : 'completed', }); if (willRetryWrapping) { unwrappedNudged = true; @@ -540,19 +542,14 @@ export async function processQuery( const names = getAllDestinations() .map((d) => d.name) .join(', '); - query.push( - `Your block was NOT delivered — task sessions deliver only via the send_message tool. ` + - `Re-send now with send_message({to: "", ...}). Your destinations: ${names}.`, - ); + query.push(buildTaskBlockNudge(taskBlocks, names)); } // A retry result (wrapping or task-block nudge) answers the SAME // user prompt — keep it queued so the retry archives against it, // not the nudge text. if (!willRetryWrapping && !willRetryTaskBlocks) archivePrompts.shift(); } - } else { - archivePrompts.shift(); - } + } else archivePrompts.shift(); } } } catch (err) { @@ -631,17 +628,22 @@ function deliverErrorResult(text: string, routing: RoutingContext): void { * The agent must always wrap output in ... * blocks, even with a single destination. Bare text is scratchpad only. */ +export interface TaskMessageBlock { + to: string; + body: string; +} + export function dispatchResultText( text: string, routing: RoutingContext, -): { sent: number; hasUnwrapped: boolean; taskBlocks: number } { +): { sent: number; hasUnwrapped: boolean; taskBlocks: TaskMessageBlock[] } { const MESSAGE_RE = /([\s\S]*?)<\/message>/g; let match: RegExpExecArray | null; let sent = 0; - // blocks left inert in a task fire — drives the same-turn + // blocks left inert in a task run — drives the same-turn // "use send_message" nudge in processQuery. - let taskBlocks = 0; + const taskBlocks: TaskMessageBlock[] = []; let lastIndex = 0; const scratchpadParts: string[] = []; @@ -657,10 +659,12 @@ export function dispatchResultText( // A final-text block here is either an echo of a tool send the // agent already made (the double-delivery class) or a send down the wrong // path — never deliver it, keep it visible in the scratchpad/run log. - if (routing.taskFire) { - log(`Task fire: block not delivered — task sessions send only via send_message`); - scratchpadParts.push(`[not delivered — task sessions send only via the send_message tool; to="${toName}"] ${body}`); - taskBlocks++; + if (routing.taskRun) { + log(`Task run: block not delivered — task sessions send only via explicit tools`); + scratchpadParts.push( + `[not delivered — task sessions send only via the send_message tool; to="${toName}"] ${body}`, + ); + taskBlocks.push({ to: toName, body }); continue; } const dest = findByName(toName); @@ -682,9 +686,9 @@ export function dispatchResultText( log(`[scratchpad] ${scratchpad.slice(0, 500)}${scratchpad.length > 500 ? '…' : ''}`); } - // In a task fire, plain final text is the NORMAL ending (it becomes the run + // In a task run, plain final text is the NORMAL ending (it becomes the run // log) — never treat it as an undelivered reply or nudge the agent to wrap it. - const hasUnwrapped = !routing.taskFire && sent === 0 && !!scratchpad; + const hasUnwrapped = !routing.taskRun && sent === 0 && !!scratchpad; if (hasUnwrapped) { log(`WARNING: agent output had no blocks — nothing was sent`); } @@ -692,23 +696,47 @@ export function dispatchResultText( } /** - * Should this task-fire result get the same-turn "your block was + * Should this task-run result get the same-turn "your block was * not delivered — use send_message" nudge? True at most once per turn * (mirrors the unwrappedNudged flag for chat turns). While true, the run-log * auto-append is DEFERRED to the retry's result so the fire logs exactly once. */ -export function shouldNudgeTaskBlocks(taskFire: boolean, taskBlocks: number, alreadyNudged: boolean): boolean { - return taskFire && taskBlocks > 0 && !alreadyNudged; +export function shouldNudgeTaskBlocks( + taskRun: boolean, + taskBlocks: TaskMessageBlock[], + alreadyNudged: boolean, +): boolean { + return taskRun && taskBlocks.length > 0 && !alreadyNudged; +} + +export function buildTaskBlockNudge(taskBlocks: TaskMessageBlock[], destinationNames: string): string { + const blocks = taskBlocks + .map( + ({ to, body }) => + `${escapePromptXml(body)}`, + ) + .join('\n'); + return ( + 'The final-output content below was not delivered from this task run:\n' + + `${blocks}\n` + + 'If and only if any of it still needs to be sent, call send_message with an explicit to destination. ' + + 'If it was already sent or no notification is required, do not send it again. ' + + `Your destinations: ${escapePromptXml(destinationNames)}. ` + + 'The original task result is already recorded in the run log; do not repeat it.' + ); +} + +function escapePromptXml(value: string): string { + return value.replace(/&/g, '&').replace(//g, '>').replace(/"/g, '"'); } /** - * Task fires: the final text IS the run log entry, unless the agent already - * logged this fire explicitly via `ncl tasks append-log` (exactly-once guard — - * old tasks whose baked-in prompt still mandates append-log don't double-log). - * Written as a `task_log` outbound row; the host appends it to the series' - * tasks/.md with its usual timestamp stamp. Never delivered to anyone. + * Task runs: the final text is the automatic run summary. Explicit + * `ncl tasks append-log` calls are additive mid-run notes. Written as a + * `task_log` outbound row; the host appends it to the series' tasks/.md + * with its usual timestamp stamp. Never delivered to anyone. */ -export function autoAppendTaskLog(text: string, turnStartSeq: number): void { +export function autoAppendTaskLog(text: string): void { // Run-log hygiene: an inert block never belongs in the log as // raw XML — replace each with its inner text, marked undelivered, so the // log stays readable prose. @@ -718,16 +746,12 @@ export function autoAppendTaskLog(text: string, turnStartSeq: number): void { ); const line = stripInternalTags(prose).replace(/\s+/g, ' ').trim().slice(0, 500); if (!line) return; - if (hasAppendLogRequestSince(turnStartSeq)) { - log('Task fire already logged via append-log — skipping final-text auto-log'); - return; - } writeMessageOut({ id: generateId(), kind: 'task_log', content: JSON.stringify({ text: line }), }); - log('Task fire run log auto-appended from final text'); + log('Task run log auto-appended from final text'); } function sendToDestination(dest: DestinationEntry, body: string, routing: RoutingContext): void { diff --git a/container/agent-runner/src/task-delivery.test.ts b/container/agent-runner/src/task-delivery.test.ts index 63e1e4145fb..e34a0c34322 100644 --- a/container/agent-runner/src/task-delivery.test.ts +++ b/container/agent-runner/src/task-delivery.test.ts @@ -1,54 +1,44 @@ /** * One-door delivery in task sessions. * - * Wiring under test (each leg goes red if its integration is deleted): - * 1. send_message with no `to` ERRORS in a task session (session_routing - * thread system:tasks:*) instead of falling back to a default target. - * 2. Final-text `` blocks are inert in a task fire — no - * outbound chat row, no "undelivered" nudge state. - * 3. The fire's final text auto-appends as a `task_log` outbound row, - * EXCEPT when the agent already ran `ncl tasks append-log` this fire. + * Every outbound tool call names its destination. In a task run, final output + * is inert delivery-wise and becomes the automatic run summary instead. */ -import { describe, it, expect, beforeEach, afterEach } from 'bun:test'; +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; -import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from './db/connection.js'; -import { getUndeliveredMessages, hasAppendLogRequestSince, maxSeq, writeMessageOut } from './db/messages-out.js'; -import { sendMessage } from './mcp-tools/core.js'; -import { autoAppendTaskLog, dispatchResultText, shouldNudgeTaskBlocks } from './poll-loop.js'; +import { closeSessionDb, getInboundDb, getOutboundDb, initTestSessionDb } from './db/connection.js'; +import { getUndeliveredMessages, writeMessageOut } from './db/messages-out.js'; +import { getTaskSeriesId } from './db/session-routing.js'; +import { sendFile, sendMessage } from './mcp-tools/core.js'; +import { autoAppendTaskLog, buildTaskBlockNudge, dispatchResultText, shouldNudgeTaskBlocks } from './poll-loop.js'; import type { RoutingContext } from './formatter.js'; -function seedSessionRouting(threadId: string | null, isTask?: 0 | 1): void { +function seedSessionRouting(channelType: string | null, platformId: string | null, threadId: string | null): void { const db = getInboundDb(); db.exec(`CREATE TABLE IF NOT EXISTS session_routing ( id INTEGER PRIMARY KEY CHECK (id = 1), - channel_type TEXT, platform_id TEXT, thread_id TEXT, - is_task INTEGER NOT NULL DEFAULT 0 + channel_type TEXT, platform_id TEXT, thread_id TEXT )`); db.prepare( - 'INSERT OR REPLACE INTO session_routing (id, channel_type, platform_id, thread_id, is_task) VALUES (1, ?, ?, ?, ?)', - ).run( - threadId ? null : 'telegram', - threadId ? null : 'telegram:123', - threadId, - isTask ?? (threadId?.startsWith('system:tasks') ? 1 : 0), - ); + 'INSERT OR REPLACE INTO session_routing (id, channel_type, platform_id, thread_id) VALUES (1, ?, ?, ?)', + ).run(channelType, platformId, threadId); } -function seedDestination(): void { +function seedDestination(name = 'family', channelType = 'telegram', platformId = 'telegram:99'): void { getInboundDb() .prepare( `INSERT INTO destinations (name, display_name, type, channel_type, platform_id, agent_group_id) - VALUES ('family', 'Family', 'channel', 'telegram', 'telegram:99', NULL)`, + VALUES (?, ?, 'channel', ?, ?, NULL)`, ) - .run(); + .run(name, name, channelType, platformId); } const taskRouting: RoutingContext = { platformId: 'ag-1', channelType: 'agent', threadId: 'system:tasks:daily-digest-a1b2', - inReplyTo: 'fire-1', - taskFire: true, + inReplyTo: 'run-1', + taskRun: true, }; beforeEach(() => { @@ -60,31 +50,58 @@ afterEach(() => { closeSessionDb(); }); -describe('send_message in a task session', () => { - it('errors without `to` — no default-destination fallback', async () => { - seedSessionRouting('system:tasks:daily-digest-a1b2'); +describe('explicit outbound destinations', () => { + it('derives task mode from the canonical per-series thread without a DB migration', () => { + seedSessionRouting(null, null, 'system:tasks:daily-digest-a1b2'); + expect(getTaskSeriesId()).toBe('daily-digest-a1b2'); - const res = (await sendMessage.handler({ text: 'hello' })) as { isError?: boolean; content: { text: string }[] }; - - expect(res.isError).toBe(true); - expect(res.content[0].text).toContain('task session'); - expect(getUndeliveredMessages()).toHaveLength(0); + seedSessionRouting('telegram', 'telegram:99', 'chat-thread'); + expect(getTaskSeriesId()).toBeNull(); }); - it('the host-stamped is_task flag alone drives the gate (thread_id NULL)', async () => { - // No thread prefix to sniff — proves the flag, not the magic string, - // is what makes the session a task session. - seedSessionRouting(null, 1); + it('requires `to` in both outbound tool schemas', () => { + expect(sendMessage.tool.inputSchema.required).toContain('to'); + expect(sendFile.tool.inputSchema.required).toContain('to'); + }); - const res = (await sendMessage.handler({ text: 'hello' })) as { isError?: boolean; content: { text: string }[] }; + it('never infers the only destination when `to` is omitted', async () => { + const messageResult = (await sendMessage.handler({ text: 'hello' })) as { + isError?: boolean; + content: { text: string }[]; + }; + const fileResult = (await sendFile.handler({ path: 'report.txt' })) as { + isError?: boolean; + content: { text: string }[]; + }; + + for (const result of [messageResult, fileResult]) { + expect(result.isError).toBe(true); + expect(result.content[0].text).toContain('to is required'); + expect(result.content[0].text).toContain('family'); + } + expect(getUndeliveredMessages()).toHaveLength(0); + }); - expect(res.isError).toBe(true); - expect(res.content[0].text).toContain('task session'); + it('rejects an unknown explicit destination without falling back', async () => { + const messageResult = (await sendMessage.handler({ to: 'missing', text: 'hello' })) as { + isError?: boolean; + content: { text: string }[]; + }; + const fileResult = (await sendFile.handler({ to: 'missing', path: 'report.txt' })) as { + isError?: boolean; + content: { text: string }[]; + }; + + for (const result of [messageResult, fileResult]) { + expect(result.isError).toBe(true); + expect(result.content[0].text).toContain('Unknown destination "missing"'); + expect(result.content[0].text).toContain('Known: family'); + } expect(getUndeliveredMessages()).toHaveLength(0); }); - it('delivers normally with an explicit `to`', async () => { - seedSessionRouting('system:tasks:daily-digest-a1b2'); + it('delivers to the explicitly named destination', async () => { + seedSessionRouting(null, null, 'system:tasks:daily-digest-a1b2'); await sendMessage.handler({ to: 'family', text: 'hello' }); @@ -93,191 +110,118 @@ describe('send_message in a task session', () => { expect(out[0].platform_id).toBe('telegram:99'); }); - it('chat sessions keep the reply-in-place default', async () => { - seedSessionRouting(null); // normal chat routing row + it('preserves the current thread for an explicitly named matching destination', async () => { + seedDestination('current-chat', 'discord', 'channel:1'); + seedSessionRouting('discord', 'channel:1', 'thread-7'); - await sendMessage.handler({ text: 'hello' }); + await sendMessage.handler({ to: 'current-chat', text: 'hello' }); const out = getUndeliveredMessages(); expect(out).toHaveLength(1); - expect(out[0].platform_id).toBe('telegram:123'); + expect(out[0].platform_id).toBe('channel:1'); + expect(out[0].thread_id).toBe('thread-7'); }); }); -describe('final-text blocks in a task fire', () => { - it('are inert — no outbound row, no undelivered flag, counted for the nudge', () => { +describe('final-output blocks in a task run', () => { + it('keeps them inert and returns their destination and content for correction', () => { const { sent, hasUnwrapped, taskBlocks } = dispatchResultText( 'digest is ready', taskRouting, ); expect(sent).toBe(0); - expect(hasUnwrapped).toBe(false); // plain text is the normal task ending — never the wrap nudge - expect(taskBlocks).toBe(1); // but the inert block IS flagged for the task nudge + expect(hasUnwrapped).toBe(false); + expect(taskBlocks).toEqual([{ to: 'family', body: 'digest is ready' }]); expect(getUndeliveredMessages()).toHaveLength(0); }); - it('still deliver in non-task sessions', () => { + it('still delivers final-output blocks in chat sessions', () => { const { sent, taskBlocks } = dispatchResultText('hi', { ...taskRouting, - taskFire: false, + taskRun: false, }); expect(sent).toBe(1); - expect(taskBlocks).toBe(0); + expect(taskBlocks).toEqual([]); expect(getUndeliveredMessages()).toHaveLength(1); }); - it('nudge fires once per turn, and only in task fires with blocks', () => { - expect(shouldNudgeTaskBlocks(true, 1, false)).toBe(true); - expect(shouldNudgeTaskBlocks(true, 1, true)).toBe(false); // already nudged this turn - expect(shouldNudgeTaskBlocks(true, 0, false)).toBe(false); // plain text is the normal ending - expect(shouldNudgeTaskBlocks(false, 1, false)).toBe(false); // chat turns use the wrap nudge + it('nudges at most once and only when a task result contains inert blocks', () => { + const blocks = [{ to: 'family', body: 'digest' }]; + expect(shouldNudgeTaskBlocks(true, blocks, false)).toBe(true); + expect(shouldNudgeTaskBlocks(true, blocks, true)).toBe(false); + expect(shouldNudgeTaskBlocks(true, [], false)).toBe(false); + expect(shouldNudgeTaskBlocks(false, blocks, false)).toBe(false); }); - it('run-log auto-append happens exactly once across a nudged turn', () => { - const start = maxSeq(); - let nudged = false; + it('shows the exact content and makes re-send conditional', () => { + const nudge = buildTaskBlockNudge([{ to: 'family', body: '3 posts & a warning' }], 'family, ops'); - // Result 1: the agent ended the fire with an inert block. - const first = dispatchResultText('digest', taskRouting); - const willRetry = shouldNudgeTaskBlocks(true, first.taskBlocks, nudged); - expect(willRetry).toBe(true); - if (!willRetry) autoAppendTaskLog('digest', start); - nudged = true; + expect(nudge).toContain('to="family"'); + expect(nudge).toContain('3 <new> posts & a warning'); + expect(nudge).toContain('If and only if'); + expect(nudge).toContain('do not send it again'); + expect(nudge).not.toContain('Re-send now'); + }); + + it('records the original task result once, not the correction retry', () => { + let nudged = false; + const original = 'digest'; + const first = dispatchResultText(original, taskRouting); + if (!nudged) autoAppendTaskLog(original); + nudged = shouldNudgeTaskBlocks(true, first.taskBlocks, nudged); - // Result 2 (post-nudge retry): plain final text → this one becomes the log. - const second = dispatchResultText('Sent the digest to family via send_message.', taskRouting); - const willRetry2 = shouldNudgeTaskBlocks(true, second.taskBlocks, nudged); - expect(willRetry2).toBe(false); - if (!willRetry2) autoAppendTaskLog('Sent the digest to family via send_message.', start); + const retry = dispatchResultText('Delivery decision handled.', taskRouting); + if (!nudged) autoAppendTaskLog('Delivery decision handled.'); + expect(shouldNudgeTaskBlocks(true, retry.taskBlocks, nudged)).toBe(false); const rows = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").all() as { content: string; }[]; expect(rows).toHaveLength(1); - expect(JSON.parse(rows[0].content).text).toBe('Sent the digest to family via send_message.'); + expect(JSON.parse(rows[0].content).text).toContain('[undelivered → family] digest'); }); }); -describe('task-fire run-log auto-append', () => { - it('writes a task_log row from the final text', () => { - const start = maxSeq(); - - autoAppendTaskLog('Checked the\nfeeds — nothing new.', start); +describe('automatic task run summary', () => { + it('writes a task_log row from final text', () => { + autoAppendTaskLog('Checked the\nfeeds — nothing new.'); const rows = getOutboundDb().prepare("SELECT kind, content FROM messages_out WHERE kind = 'task_log'").all() as { kind: string; content: string; }[]; expect(rows).toHaveLength(1); - expect(JSON.parse(rows[0].content).text).toBe('Checked the feeds — nothing new.'); // whitespace collapsed + expect(JSON.parse(rows[0].content).text).toBe('Checked the feeds — nothing new.'); }); - it('strips blocks — logs their inner text marked undelivered, never raw XML', () => { - const start = maxSeq(); - - autoAppendTaskLog('Digest done. 3 new posts today See you tomorrow.', start); + it('marks legacy final-output blocks undelivered and never stores raw XML', () => { + autoAppendTaskLog('Digest done. 3 new posts today See you tomorrow.'); - const rows = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").all() as { + const row = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").get() as { content: string; - }[]; - expect(rows).toHaveLength(1); - const line = JSON.parse(rows[0].content).text as string; + }; + const line = JSON.parse(row.content).text as string; expect(line).not.toContain(' { - const start = maxSeq(); - // The ncl binary writes each CLI call as a cli_request system row. + it('is additive to an explicit append-log request', () => { writeMessageOut({ - id: 'cli-1', - kind: 'system', - content: JSON.stringify({ action: 'cli_request', requestId: 'cli-1', command: 'tasks-append-log', args: { msg: 'done' } }), - }); - expect(hasAppendLogRequestSince(start)).toBe(true); - - autoAppendTaskLog('final text', start); - - const rows = getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all(); - expect(rows).toHaveLength(0); - }); - - it('a positional append-log invocation (dash-joined command) also suppresses', () => { - const start = maxSeq(); - // `ncl tasks append-log "did the thing"` → command 'tasks-append-log-did-the-thing' - writeMessageOut({ - id: 'cli-pos', + id: 'cli-progress', kind: 'system', content: JSON.stringify({ action: 'cli_request', - requestId: 'cli-pos', - command: 'tasks-append-log-did-the-thing', - args: {}, - }), - }); - - expect(hasAppendLogRequestSince(start)).toBe(true); - - autoAppendTaskLog('final text', start); - expect(getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all()).toHaveLength(0); - }); - - it('a DEFINITIVELY failed append-log (response ok:false) does not suppress', () => { - const start = maxSeq(); - writeMessageOut({ - id: 'cli-fail', - kind: 'system', - content: JSON.stringify({ action: 'cli_request', requestId: 'cli-fail', command: 'tasks-append-log', args: {} }), - }); - // Host response frame lands in inbound messages_in with ok:false. - getInboundDb() - .prepare("INSERT INTO messages_in (id, seq, kind, timestamp, content) VALUES (?, ?, 'system', datetime('now'), ?)") - .run( - 'resp-fail', - 1000, - JSON.stringify({ - requestId: 'cli-fail', - frame: { id: 'cli-fail', ok: false, error: { code: 'invalid-args', message: 'bad series' } }, - }), - ); - - expect(hasAppendLogRequestSince(start)).toBe(false); - - autoAppendTaskLog('final text', start); - expect(getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all()).toHaveLength(1); - }); - - it('an append-log with a still-pending response suppresses (no double-log on the success path)', () => { - const start = maxSeq(); - writeMessageOut({ - id: 'cli-pending', - kind: 'system', - content: JSON.stringify({ - action: 'cli_request', - requestId: 'cli-pending', + requestId: 'cli-progress', command: 'tasks-append-log', - args: {}, + args: { msg: 'progress note' }, }), }); - // No response row in inbound yet. - expect(hasAppendLogRequestSince(start)).toBe(true); - }); + autoAppendTaskLog('final summary'); - it('append-log from BEFORE the fire does not suppress', () => { - writeMessageOut({ - id: 'cli-old', - kind: 'system', - content: JSON.stringify({ action: 'cli_request', requestId: 'cli-old', command: 'tasks-append-log', args: {} }), - }); - const start = maxSeq(); // watermark taken after the old call - - autoAppendTaskLog('final text', start); - - const rows = getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all(); - expect(rows).toHaveLength(1); + expect(getOutboundDb().prepare("SELECT 1 FROM messages_out WHERE kind = 'task_log'").all()).toHaveLength(1); }); }); diff --git a/src/cli/resources/destinations.ts b/src/cli/resources/destinations.ts index 87e39fc9dd5..dbb5e01379a 100644 --- a/src/cli/resources/destinations.ts +++ b/src/cli/resources/destinations.ts @@ -45,7 +45,7 @@ registerResource({ name: 'local_name', type: 'string', description: - 'Name the agent uses to address this target (e.g. ). Unique per agent. Lowercase, dash-separated.', + 'Name the agent uses to address this target (e.g. send_message({ to: "local_name", ... })). Unique per agent. Lowercase, dash-separated.', }, { name: 'target_type', diff --git a/src/cli/resources/tasks.test.ts b/src/cli/resources/tasks.test.ts index bd0a5cd6e33..98cbcf44bd7 100644 --- a/src/cli/resources/tasks.test.ts +++ b/src/cli/resources/tasks.test.ts @@ -105,8 +105,8 @@ describe('tasks CLI resource', () => { 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 + expect(content.prompt).toBe('send a briefing'); + expect(content.prompt).not.toContain('Task delivery contract'); systemDb.close(); }); diff --git a/src/cli/resources/tasks.ts b/src/cli/resources/tasks.ts index 4a7676a4419..c893416a49f 100644 --- a/src/cli/resources/tasks.ts +++ b/src/cli/resources/tasks.ts @@ -281,22 +281,17 @@ function createTask(args: Record, ctx: CallerContext) { 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. + // Each series runs in its own isolated session. Delivery and run-log + // instructions are injected by that session's runtime prompt rather than + // baked into persisted task content, so the contract can evolve in code. const { session } = resolveTaskSession(group, id); - const promptWithLog = - `${prompt}\n\n` + - `[Task delivery contract:\n` + - `• MESSAGE (only if the task asks you to report/notify): use send_message({ to: "name", … }) with an explicit destination — that tool call is the ONLY thing the user receives. This run has no chat attached: final text and blocks are NOT delivered here.\n` + - `• RUN LOG (automatic): your final text is recorded verbatim in tasks/${id}.md — end the run with a concrete work-log line: what you did and WHY (a no-op run still ends with why nothing was needed; name any files you wrote). Not a greeting, not a copy of the message you sent. For extra mid-run notes use \`ncl tasks append-log --msg "…"\` — if you do, your final text is not auto-logged. Do NOT edit tasks/${id}.md by hand; the log 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 }), + content: JSON.stringify({ prompt, script, originSessionId }), }); return selectTask(db, id); }); @@ -308,7 +303,7 @@ function createTask(args: Record, ctx: CallerContext) { * Append one host-timestamped line to a task's run log * (`//tasks/.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 + * can see when and why each run happened. Inside a task run the series is derived from * the caller's own task session, so the agent supplies only --msg. */ function appendTaskLog( @@ -648,9 +643,9 @@ registerResource({ 'append-log': { access: 'open', description: - 'Append a one-line note to a task run log (tasks/.md).\n\nOptional: a task fire auto-logs its final text, so most runs need no explicit call — use this for mid-run notes (calling it suppresses the final-text auto-log for that fire). The 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.', + 'Append a one-line note to a task run log (tasks/.md).\n\nOptional: every task run auto-logs its final text, so use this only for additive mid-run notes. The host stamps the UTC timestamp; you supply --msg. This is a LOG ENTRY, not a message — it sends nothing to anyone. Inside a task run --id is auto-derived from your session.', 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"`, + `# Inside a task run (--id auto-derived) — optional progress note:\nncl tasks append-log --msg "one feed returned 403; continuing with the remaining feeds"`, ], args: [ { @@ -663,7 +658,7 @@ registerResource({ { name: 'id', type: 'string', - description: 'Task series id. Auto-derived when called from inside a task fire; required otherwise.', + description: 'Task series id. Auto-derived when called from inside a task run; required otherwise.', }, { name: 'group', diff --git a/src/container-runner.ts b/src/container-runner.ts index 7330a4c52ab..ff8f6ad0789 100644 --- a/src/container-runner.ts +++ b/src/container-runner.ts @@ -117,7 +117,7 @@ async function spawnContainer(session: Session): Promise { return; } - // Refresh the destination map and default reply routing so any admin + // Refresh the destination map and current-thread routing so any admin // changes take effect on wake. Destinations come from the agent-to-agent // module — skip when the module isn't installed (table absent). if (hasTable(getDb(), 'agent_destinations')) { diff --git a/src/db/schema.ts b/src/db/schema.ts index 53daa983fbc..33db4aaad6c 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -210,20 +210,16 @@ CREATE TABLE IF NOT EXISTS destinations ( agent_group_id TEXT -- for type='agent' ); --- Default reply routing for this session. Single-row table (id=1). +-- Current chat/thread routing for this session. Single-row table (id=1). -- Host overwrites on every container wake from the session's messaging_group -- and thread_id. Container reads it in send_message / ask_user_question to --- default the channel/thread of outbound messages when the agent doesn't --- specify an explicit destination. +-- preserve the thread when an explicitly named destination is the current +-- conversation, and for interactive-question response matching. CREATE TABLE IF NOT EXISTS session_routing ( id INTEGER PRIMARY KEY CHECK (id = 1), channel_type TEXT, platform_id TEXT, - thread_id TEXT, - -- 1 = this session is a task-series session (thread system:tasks:*). - -- Stamped by the host so the container never has to sniff the magic - -- thread prefix (which fails open when thread_id is absent/renamed). - is_task INTEGER NOT NULL DEFAULT 0 + thread_id TEXT ); `; diff --git a/src/db/session-db.test.ts b/src/db/session-db.test.ts index f1c2ad3fa60..043addb1edf 100644 --- a/src/db/session-db.test.ts +++ b/src/db/session-db.test.ts @@ -10,13 +10,7 @@ import fs from 'fs'; import path from 'path'; import { describe, it, expect, afterEach } from 'vitest'; -import { - ensureSchema, - getInboundSourceSessionId, - migrateMessagesInTable, - syncProcessingAcks, - upsertSessionRouting, -} 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'); @@ -99,49 +93,6 @@ describe('migrateMessagesInTable', () => { }); }); -describe('session_routing is_task stamp', () => { - it('upsert writes is_task, migrating a legacy table without the column', () => { - if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); - fs.mkdirSync(TEST_DIR, { recursive: true }); - - // Legacy session_routing (pre-is_task). - const db = new Database(DB_PATH); - db.exec(`CREATE TABLE session_routing ( - id INTEGER PRIMARY KEY CHECK (id = 1), - channel_type TEXT, platform_id TEXT, thread_id TEXT - )`); - - upsertSessionRouting(db, { - channel_type: null, - platform_id: null, - thread_id: 'system:tasks:daily-abc1', - is_task: 1, - }); - // Idempotent re-run flips it back off (host overwrites on every wake). - upsertSessionRouting(db, { channel_type: 'telegram', platform_id: 't:1', thread_id: null, is_task: 0 }); - - const row = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; - expect(row.is_task).toBe(0); - - upsertSessionRouting(db, { channel_type: null, platform_id: null, thread_id: null, is_task: 1 }); - const row2 = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; - expect(row2.is_task).toBe(1); - db.close(); - }); - - it('fresh inbound schema already carries is_task', () => { - if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); - fs.mkdirSync(TEST_DIR, { recursive: true }); - ensureSchema(DB_PATH, 'inbound'); - const db = new Database(DB_PATH); - const cols = (db.prepare("PRAGMA table_info('session_routing')").all() as Array<{ name: string }>).map( - (c) => c.name, - ); - expect(cols).toContain('is_task'); - db.close(); - }); -}); - describe('syncProcessingAcks — script-skip counter', () => { function freshPair() { if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); diff --git a/src/db/session-db.ts b/src/db/session-db.ts index ff9ce796c00..3424fce659e 100644 --- a/src/db/session-db.ts +++ b/src/db/session-db.ts @@ -40,31 +40,17 @@ export function openOutboundDbRw(dbPath: string): Database.Database { return db; } -/** - * Lazy, on-open migration for session_routing (mirrors migrateMessagesInTable): - * pre-existing session DBs lack the is_task column. Idempotent; no-op when the - * table itself is missing (older DBs error on upsert exactly as before). - */ -export function migrateSessionRoutingTable(db: Database.Database): void { - const cols = (db.prepare("PRAGMA table_info('session_routing')").all() as Array<{ name: string }>).map((c) => c.name); - if (cols.length > 0 && !cols.includes('is_task')) { - db.prepare('ALTER TABLE session_routing ADD COLUMN is_task INTEGER NOT NULL DEFAULT 0').run(); - } -} - export function upsertSessionRouting( db: Database.Database, - routing: { channel_type: string | null; platform_id: string | null; thread_id: string | null; is_task: 0 | 1 }, + routing: { channel_type: string | null; platform_id: string | null; thread_id: string | null }, ): void { - migrateSessionRoutingTable(db); db.prepare( - `INSERT INTO session_routing (id, channel_type, platform_id, thread_id, is_task) - VALUES (1, @channel_type, @platform_id, @thread_id, @is_task) + `INSERT INTO session_routing (id, channel_type, platform_id, thread_id) + VALUES (1, @channel_type, @platform_id, @thread_id) ON CONFLICT(id) DO UPDATE SET channel_type = excluded.channel_type, platform_id = excluded.platform_id, - thread_id = excluded.thread_id, - is_task = excluded.is_task`, + thread_id = excluded.thread_id`, ).run(routing); } diff --git a/src/delivery.ts b/src/delivery.ts index 474c02fb1c6..7a8cacf939e 100644 --- a/src/delivery.ts +++ b/src/delivery.ts @@ -267,7 +267,7 @@ async function deliverMessage( return; } - // Task-fire run log: the runner mirrors a fire's final text here (one-door + // Task-run log: the runner mirrors a run's final text here (one-door // delivery — final text never reaches a channel; the send_message tool is // the only delivery path from a task session). Append to the series log, // never deliver. The caller marks it delivered so it isn't retried. diff --git a/src/host-core.test.ts b/src/host-core.test.ts index 81b3635e919..5bee875f4fc 100644 --- a/src/host-core.test.ts +++ b/src/host-core.test.ts @@ -19,7 +19,6 @@ import { } from './db/index.js'; import { resolveSession, - resolveTaskSession, writeSessionMessage, writeSessionRouting, initSessionFolder, @@ -954,32 +953,6 @@ describe('writeSessionRouting', () => { expect(row!.thread_id).toBeNull(); }); - it('stamps is_task=1 for a task-series session and 0 for chat sessions', () => { - createAgentGroup({ - id: 'ag-1', - name: 'Agent', - folder: 'agent', - agent_provider: null, - created_at: now(), - }); - - const { session: taskSession } = resolveTaskSession('ag-1', 'daily-abc1'); - writeSessionRouting('ag-1', taskSession.id); - - let db = new Database(inboundDbPath('ag-1', taskSession.id)); - const taskRow = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; - db.close(); - expect(taskRow.is_task).toBe(1); - - const { session: chatSession } = resolveSession('ag-1', null, null, 'agent-shared'); - writeSessionRouting('ag-1', chatSession.id); - - db = new Database(inboundDbPath('ag-1', chatSession.id)); - const chatRow = db.prepare('SELECT is_task FROM session_routing WHERE id = 1').get() as { is_task: number }; - db.close(); - expect(chatRow.is_task).toBe(0); - }); - it('includes thread_id from per-thread session', () => { createAgentGroup({ id: 'ag-1', diff --git a/src/modules/agent-to-agent/create-agent.ts b/src/modules/agent-to-agent/create-agent.ts index b8044da390d..8dd3a2d54dc 100644 --- a/src/modules/agent-to-agent/create-agent.ts +++ b/src/modules/agent-to-agent/create-agent.ts @@ -206,6 +206,6 @@ async function performCreateAgent( // tries to send to the newly-created child. writeDestinations(session.agent_group_id, session.id); - notify(`Agent "${localName}" created. You can now message it with ....`); + notify(`Agent "${localName}" created. You can now message it with send_message({ to: "${localName}", ... }).`); log.info('Agent group created', { agentGroupId, name, localName, folder, parent: sourceGroup.id }); } diff --git a/src/modules/scheduling/run-log.ts b/src/modules/scheduling/run-log.ts index 3684a801b66..a4657687560 100644 --- a/src/modules/scheduling/run-log.ts +++ b/src/modules/scheduling/run-log.ts @@ -4,7 +4,7 @@ * * Two writers, one format: * - `ncl tasks append-log` (agent's explicit mid-run/work-log entry) - * - the `task_log` outbound row a task fire's final text produces + * - the `task_log` outbound row a task run's final text produces * (container/agent-runner poll-loop auto-append; delivery.ts routes it here) */ import fs from 'fs'; diff --git a/src/session-manager.ts b/src/session-manager.ts index 62dff1b4128..adb0418150c 100644 --- a/src/session-manager.ts +++ b/src/session-manager.ts @@ -26,7 +26,6 @@ import { findSessionByAgentGroup, findSessionForAgent, getSession, - isTaskThread, taskThreadId, updateSession, } from './db/sessions.js'; @@ -161,10 +160,10 @@ export function initSessionFolder(agentGroupId: string, sessionId: string): void } /** - * Write the default reply routing for a session into its inbound.db. + * Write the current chat/thread routing for a session into its inbound.db. * - * The container reads this as the default (channel_type, platform_id, thread_id) - * for outbound messages when the agent doesn't specify an explicit destination. + * The container uses this to preserve thread_id when an explicitly named + * destination resolves to the conversation this session is bound to. * Derived from session.messaging_group_id → messaging_groups row + session.thread_id. * * Called on every container wake alongside the agent-to-agent module's @@ -194,10 +193,6 @@ export function writeSessionRouting(agentGroupId: string, sessionId: string): vo channel_type: channelType, platform_id: platformId, thread_id: session.thread_id, - // Stamp task-ness explicitly so the container can gate one-door - // delivery on a host-asserted flag instead of sniffing the thread - // prefix (fail-closed at the source of truth). - is_task: isTaskThread(session.thread_id) ? 1 : 0, }); } finally { db.close(); From 3d4b349ba9cecb1cdff207a97227877d57974d38 Mon Sep 17 00:00:00 2001 From: Omri Maya Date: Wed, 8 Jul 2026 12:46:23 +0300 Subject: [PATCH 3/7] test(tasks): cover one-door task turns --- container/agent-runner/src/poll-loop.test.ts | 86 ++++++++++++++++++++ container/agent-runner/src/poll-loop.ts | 3 +- 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/container/agent-runner/src/poll-loop.test.ts b/container/agent-runner/src/poll-loop.test.ts index 023c59e23af..fa1c6279069 100644 --- a/container/agent-runner/src/poll-loop.test.ts +++ b/container/agent-runner/src/poll-loop.test.ts @@ -454,3 +454,89 @@ describe('isCorruptionError', () => { expect(isCorruptionError('')).toBe(false); }); }); + +// --- Task-run turn wiring: the REAL processQuery path (one-door) --- +// These drive the actual call sites (autoAppendTaskLog at result-handling, +// shouldNudgeTaskBlocks gating, and follow-up turn reset). Deleting the wiring +// — not just the helpers — goes red here. + +const TASK_ROUTING = { + platformId: null, + channelType: null, + threadId: 'system:tasks:ser-1', + inReplyTo: 't1', + taskRun: true, +}; + +function taskLogRows(): Array<{ text: string }> { + return ( + getOutboundDb() + .prepare("SELECT content FROM messages_out WHERE kind = 'task_log' ORDER BY seq") + .all() as Array<{ content: string }> + ).map((r) => JSON.parse(r.content) as { text: string }); +} + +describe('task-run turn wiring (real processQuery)', () => { + it('auto-appends the final text as a task_log row', async () => { + async function* events(): AsyncGenerator { + yield { type: 'init', continuation: 's1' }; + yield { type: 'result', text: 'checked feeds — nothing new' }; + } + const query: AgentQuery = { push: () => {}, end: () => {}, events: events(), abort: () => {} }; + + await processQuery(query, TASK_ROUTING, ['t1'], 'claude', undefined, 'prompt', undefined); + + const logs = taskLogRows(); + expect(logs).toHaveLength(1); + expect(logs[0].text).toBe('checked feeds — nothing new'); + // and nothing was delivered as chat + expect(getUndeliveredMessages().filter((m) => m.kind === 'chat')).toHaveLength(0); + }); + + it('logs and conditionally nudges a second task run in the same open query', async () => { + const pushes: string[] = []; + + async function* events(): AsyncGenerator { + yield { type: 'init', continuation: 's1' }; + // Turn 1 uses the legacy wrong door and consumes its one correction. + yield { type: 'result', text: 'fire one result' }; + yield { type: 'result', text: 'first delivery decision handled' }; + + // A SECOND task run lands while the query is open — the follow-up poller + // pushes it and must reset the per-turn correction state. + insertMessage('t2', 'task', { prompt: 'fire two' }); + const deadline = Date.now() + 5000; + while (!pushes.some((p) => p.includes('fire two')) && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 50)); + } + + // Turn 2 repeats the mistake. This receives a second independent nudge + // only if the follow-up path reset taskBlockNudged. + yield { type: 'result', text: 'fire two result' }; + yield { type: 'result', text: 'second delivery decision handled' }; + } + + const query: AgentQuery = { + push: (m: string) => { + pushes.push(m); + }, + end: () => {}, + events: events(), + abort: () => {}, + }; + + await processQuery(query, TASK_ROUTING, ['t1'], 'claude', undefined, 'prompt', undefined); + + const nudges = pushes.filter((p) => p.includes('If and only if')); + expect(nudges).toHaveLength(2); + expect(nudges[0]).toContain('fire one result'); + expect(nudges[1]).toContain('fire two result'); + + const logs = taskLogRows().map((l) => l.text); + expect(logs).toHaveLength(2); + expect(logs[0]).toContain('[undelivered → local-cli] fire one result'); + expect(logs[1]).toContain('[undelivered → local-cli] fire two result'); + expect(logs).not.toContain('first delivery decision handled'); + expect(logs).not.toContain('second delivery decision handled'); + }); +}); diff --git a/container/agent-runner/src/poll-loop.ts b/container/agent-runner/src/poll-loop.ts index 7e304938f2f..c823cca0c42 100644 --- a/container/agent-runner/src/poll-loop.ts +++ b/container/agent-runner/src/poll-loop.ts @@ -698,8 +698,7 @@ export function dispatchResultText( /** * Should this task-run result get the same-turn "your block was * not delivered — use send_message" nudge? True at most once per turn - * (mirrors the unwrappedNudged flag for chat turns). While true, the run-log - * auto-append is DEFERRED to the retry's result so the fire logs exactly once. + * (mirrors the unwrappedNudged flag for chat turns). */ export function shouldNudgeTaskBlocks( taskRun: boolean, From 379fd28c24f6ed66393a26281e05be52131a2caa Mon Sep 17 00:00:00 2001 From: Omri Maya Date: Sat, 11 Jul 2026 21:15:57 +0300 Subject: [PATCH 4/7] fix(tasks): tighten the task-mode delivery instruction per review Co-Authored-By: Claude Fable 5 --- container/agent-runner/src/destinations.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/container/agent-runner/src/destinations.ts b/container/agent-runner/src/destinations.ts index 6c8badac44a..74a5efbe217 100644 --- a/container/agent-runner/src/destinations.ts +++ b/container/agent-runner/src/destinations.ts @@ -116,7 +116,7 @@ function buildDestinationsSection(mode: SessionMode): string { lines.push( 'This is an isolated task run with no attached chat. Only notify someone when the task asks you to. For a user-visible message, call `send_message({ to: "name", text: "..." })`; for a file, call `send_file` with `to`. Always pass the explicit named destination.', '', - `Your final response is not delivered. End with a concise work-log summary of what happened and why; explain a no-op and name files you changed. It is recorded automatically in \`tasks/${mode.taskId}.md\`. Read that file when you need context from earlier runs. Use \`ncl tasks append-log --msg "…"\` only for optional mid-run notes.`, + `Your final output is not sent to the user. End with a concise work-log summary. It is recorded automatically in \`tasks/${mode.taskId}.md\`. Read that file when you need context from earlier runs. Use \`ncl tasks append-log --msg "…"\` only for optional mid-run notes.`, ); return lines.join('\n'); } From a3be5d062bb18ba02f4f01986cd4da221cd11c00 Mon Sep 17 00:00:00 2001 From: gavrielc Date: Mon, 13 Jul 2026 16:00:19 +0300 Subject: [PATCH 5/7] Apply suggestion from @gavrielc --- container/agent-runner/src/mcp-tools/scheduling.instructions.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/container/agent-runner/src/mcp-tools/scheduling.instructions.md b/container/agent-runner/src/mcp-tools/scheduling.instructions.md index 033c4da9d6b..670b8854ac5 100644 --- a/container/agent-runner/src/mcp-tools/scheduling.instructions.md +++ b/container/agent-runner/src/mcp-tools/scheduling.instructions.md @@ -7,7 +7,7 @@ Pass `--name ""` on create to get a readable task id (e.g. `--name Common commands: ```bash -ncl tasks create --name "ping" --prompt "Remind me to call Dana" --process-after "tomorrow 18:00" +ncl tasks create --name "ping" --prompt "Remind the user 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) From fd55cc4a423d8d36d803f11b580be5d7232e6982 Mon Sep 17 00:00:00 2001 From: gavrielc Date: Mon, 13 Jul 2026 16:04:14 +0300 Subject: [PATCH 6/7] Apply suggestion from @gavrielc --- container/agent-runner/src/compact-instructions.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/container/agent-runner/src/compact-instructions.ts b/container/agent-runner/src/compact-instructions.ts index 38bcf37376a..4a3cf1eee39 100644 --- a/container/agent-runner/src/compact-instructions.ts +++ b/container/agent-runner/src/compact-instructions.ts @@ -15,7 +15,7 @@ import { getTaskSeriesId } from './db/session-routing.js'; export function buildCompactInstructions(names: string[], taskId: string | null): string { const deliveryReminder = taskId ? [ - ' "This is an isolated task run. Send a user-visible message only with send_message and an explicit to destination.', + ' "This is an isolated task run. If you need to send the user a message, use send_message with an explicit to destination.', ` Final output is not delivered; it becomes the automatic summary in tasks/${taskId}.md.`, ` Available destinations: ${formatDestinationNames(names)}."`, ] From 89c3a9bdd81ca81c8a3c27cbc4dd6bbf84767a7b Mon Sep 17 00:00:00 2001 From: gavrielc Date: Mon, 13 Jul 2026 16:06:36 +0300 Subject: [PATCH 7/7] Apply suggestion from @gavrielc --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 87770ad9e6d..74fb887b36f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ All notable changes to NanoClaw will be documented in this file. ## [Unreleased] -- [BREAKING] **Explicit destinations and one-door task delivery.** Every `send_message` and `send_file` call now requires a named `to` destination; there is no single-destination or reply-in-place shortcut. In task sessions, only explicitly addressed tool calls deliver, while final output becomes the automatic run summary and `ncl tasks append-log` adds optional progress notes. Existing task DBs need no schema migration; legacy generated task instructions are normalized when read. **Migration:** rebuild the agent image, restart NanoClaw, and update custom instructions that omit `to`. +- [BREAKING] **Explicit destinations and one-door task delivery.** Every `send_message` and `send_file` call now requires a named `to` destination; there is no single-destination or reply-in-place shortcut. In task sessions, only explicitly addressed tool calls deliver, while final output becomes the automatic run summary and `ncl tasks append-log` adds optional progress notes. Existing task DBs need no schema migration; legacy generated task instructions are normalized when read. **Migration:** rebuild the agent image, restart NanoClaw, update custom instructions that omit `to`, and clear or compact existing sessions. - [BREAKING] **`whatsapp-formatting` and `slack-formatting` container skills moved from trunk to the `channels` branch.** They now install with their channel — `/add-whatsapp` / `/add-slack` and the setup installers copy them in — so installs without those channels stop carrying channel-specific formatting instructions in every agent's context. **Migration — only if this install has the channel wired** (check `src/channels/whatsapp.ts` / `src/channels/slack.ts`): updating removes both skills from the working tree — WhatsApp agents lose the formatting fragment from their composed CLAUDE.md on next spawn, Slack agents lose the mrkdwn skill from `~/.claude/skills`. Re-run the matching skill, `/add-whatsapp` or `/add-slack` (idempotent), to restore. Installs without the channel need nothing — do NOT run the add-skill just in case; it installs the full channel adapter. - **Pre-task script failures back their series off instead of spinning.** A `--script` that errors lands the occurrence as a failed run (`script-skip:error` ack → `failed` status); recurrence reads the series' trailing failed streak and re-arms at `max(cron next, now + 2·2^(n−1) min, cap 60)`; after 8 consecutive failures the series is auto-paused with a host-written note in its run log (`ncl tasks resume` revives it). A deliberate `wakeAgent:false` gate is a normal run and never backs off. - [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).