diff --git a/CHANGELOG.md b/CHANGELOG.md index 640c05f903a..3e4a1f2899c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,10 +4,11 @@ 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, update custom instructions that omit `to`, and clear or compact existing sessions. - **The guard seam.** Every privileged action crossing the container or channel boundary now passes one decision function — `guard()` in `src/guard/` — before it executes: `allow`, `hold` (the existing approval flows), or `deny`. Today's checks are preserved verbatim as each action's decision; ncl commands and delivery actions cannot register without a guard, and approved replays re-enter carrying the approval row as a **grant** (the forgeable `approved: true` boolean is deleted) with the checks re-run live. Two deliberate outcome changes: **(a)** approving a held a2a message after its destination was revoked no longer delivers it — the requester is told "approved, but not delivered" and the host logs a warning; **(b)** a forged, already-consumed, or mismatched grant refuses the replay instead of executing. - **Delivery-registry hardening.** Re-registering a guard-wrapped delivery action *without* a guard spec now throws instead of silently disarming the guard. - [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. 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. +- **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/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..4a3cf1eee39 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. 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)}."`, + ] + : [ + ' "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 6f834fbee36..6c5c5b2d7bd 100644 --- a/container/agent-runner/src/db/messages-out.ts +++ b/container/agent-runner/src/db/messages-out.ts @@ -142,22 +142,3 @@ export function getUndeliveredMessages(): MessageOutRow[] { ) .all() as MessageOutRow[]; } - -/** - * True if a deliberate send with this exact destination + text already exists - * (an MCP send_message row from the current turn). Used by the task-fire - * final-text dispatcher to drop the turn-final echo of a send the - * agent already made — the dedup happens where the duplication originates. - */ -export function hasIdenticalSend(platformId: string, channelType: string, text: string): boolean { - const row = getOutboundDb() - .prepare( - `SELECT 1 FROM messages_out - WHERE platform_id = $platform_id AND channel_type = $channel_type - AND (in_reply_to IS NULL OR in_reply_to = '') - AND json_extract(content, '$.text') = $text - LIMIT 1`, - ) - .get({ $platform_id: platformId, $channel_type: channelType, $text: text }); - return row != null; -} diff --git a/container/agent-runner/src/db/session-routing.ts b/container/agent-runner/src/db/session-routing.ts index 94abca63100..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'; @@ -19,12 +16,20 @@ export interface SessionRouting { 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') - .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 { - // 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 }; } + +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 26dd99c6723..9a65fb31f2c 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, formatLocalTime } 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 35a1f8f88df..232ab9108ad 100644 --- a/container/agent-runner/src/formatter.ts +++ b/container/agent-runner/src/formatter.ts @@ -98,11 +98,10 @@ 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. */ - taskFire: boolean; + /** 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. */ + taskRun: boolean; } /** @@ -116,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'), }; } @@ -209,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 452ce2d8dd5..b240e55d465 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 @@ -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 local 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; the host stamps the local time. +# 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 b9b46e087e1..61bd5579c45 100644 --- a/container/agent-runner/src/mcp-tools/core.ts +++ b/container/agent-runner/src/mcp-tools/core.ts @@ -41,39 +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) { - // Default: reply to whatever thread/channel this session is bound to. - const session = getSessionRouting(); - 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') { @@ -95,24 +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`.', + 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(); @@ -134,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 386ceae8231..670b8854ac5 100644 --- a/container/agent-runner/src/mcp-tools/scheduling.instructions.md +++ b/container/agent-runner/src/mcp-tools/scheduling.instructions.md @@ -1,13 +1,13 @@ ## 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. 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-`. 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) 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 347ad8f3c42..c823cca0c42 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 { hasIdenticalSend, 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,6 +346,9 @@ export async function processQuery( let queryContinuation: string | undefined; let done = false; let unwrappedNudged = false; + // 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 // oldest unanswered prompt, except a wrapping-retry result, which answers // the same prompt again. Unused (and unmaintained) when the provider @@ -423,6 +432,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); @@ -487,8 +497,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.taskRun, taskBlocks, taskBlockNudged); + // One-door task delivery: the final text becomes the run log entry + // 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 @@ -507,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; @@ -520,13 +537,19 @@ 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(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) { @@ -605,11 +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. */ -function dispatchResultText(text: string, routing: RoutingContext): { sent: number; hasUnwrapped: boolean } { +export interface TaskMessageBlock { + to: string; + body: string; +} + +export function dispatchResultText( + text: string, + routing: RoutingContext, +): { 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 run — drives the same-turn + // "use send_message" nudge in processQuery. + const taskBlocks: TaskMessageBlock[] = []; let lastIndex = 0; const scratchpadParts: string[] = []; @@ -621,6 +655,18 @@ 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.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); if (!dest) { log(`Unknown destination in , dropping block`); @@ -640,25 +686,76 @@ 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 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.taskRun && 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-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). + */ +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 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): 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; + writeMessageOut({ + id: generateId(), + kind: 'task_log', + content: JSON.stringify({ text: line }), + }); + log('Task 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 +763,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..e34a0c34322 --- /dev/null +++ b/container/agent-runner/src/task-delivery.test.ts @@ -0,0 +1,227 @@ +/** + * One-door delivery in task sessions. + * + * 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 { afterEach, beforeEach, describe, expect, it } from 'bun:test'; + +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(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 + )`); + db.prepare( + 'INSERT OR REPLACE INTO session_routing (id, channel_type, platform_id, thread_id) VALUES (1, ?, ?, ?)', + ).run(channelType, platformId, threadId); +} + +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 (?, ?, 'channel', ?, ?, NULL)`, + ) + .run(name, name, channelType, platformId); +} + +const taskRouting: RoutingContext = { + platformId: 'ag-1', + channelType: 'agent', + threadId: 'system:tasks:daily-digest-a1b2', + inReplyTo: 'run-1', + taskRun: true, +}; + +beforeEach(() => { + initTestSessionDb(); + seedDestination(); +}); + +afterEach(() => { + closeSessionDb(); +}); + +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'); + + seedSessionRouting('telegram', 'telegram:99', 'chat-thread'); + expect(getTaskSeriesId()).toBeNull(); + }); + + it('requires `to` in both outbound tool schemas', () => { + expect(sendMessage.tool.inputSchema.required).toContain('to'); + expect(sendFile.tool.inputSchema.required).toContain('to'); + }); + + 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); + }); + + 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 to the explicitly named destination', async () => { + seedSessionRouting(null, null, '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('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({ to: 'current-chat', text: 'hello' }); + + const out = getUndeliveredMessages(); + expect(out).toHaveLength(1); + expect(out[0].platform_id).toBe('channel:1'); + expect(out[0].thread_id).toBe('thread-7'); + }); +}); + +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); + expect(taskBlocks).toEqual([{ to: 'family', body: 'digest is ready' }]); + expect(getUndeliveredMessages()).toHaveLength(0); + }); + + it('still delivers final-output blocks in chat sessions', () => { + const { sent, taskBlocks } = dispatchResultText('hi', { + ...taskRouting, + taskRun: false, + }); + + expect(sent).toBe(1); + expect(taskBlocks).toEqual([]); + expect(getUndeliveredMessages()).toHaveLength(1); + }); + + 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('shows the exact content and makes re-send conditional', () => { + const nudge = buildTaskBlockNudge([{ to: 'family', body: '3 posts & a warning' }], 'family, ops'); + + 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); + + 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).toContain('[undelivered → family] digest'); + }); +}); + +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.'); + }); + + it('marks legacy final-output blocks undelivered and never stores raw XML', () => { + autoAppendTaskLog('Digest done. 3 new posts today See you tomorrow.'); + + const row = getOutboundDb().prepare("SELECT content FROM messages_out WHERE kind = 'task_log'").get() as { + content: string; + }; + const line = JSON.parse(row.content).text as string; + expect(line).not.toContain(' { + writeMessageOut({ + id: 'cli-progress', + kind: 'system', + content: JSON.stringify({ + action: 'cli_request', + requestId: 'cli-progress', + command: 'tasks-append-log', + args: { msg: 'progress note' }, + }), + }); + + autoAppendTaskLog('final summary'); + + 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 f440729146b..6be522f3e1d 100644 --- a/src/cli/resources/destinations.ts +++ b/src/cli/resources/destinations.ts @@ -47,7 +47,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 14091a5851e..6c24be4ad66 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 35b1713550a..3038afb96f9 100644 --- a/src/cli/resources/tasks.ts +++ b/src/cli/resources/tasks.ts @@ -24,8 +24,9 @@ import { type TaskUpdate, } from '../../modules/scheduling/db.js'; import { inboundDbPath, resolveTaskSession, withInboundDb } from '../../session-manager.js'; -import { formatLocalStamp, parseZonedToUtc } from '../../timezone.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'; @@ -280,24 +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` + - `[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 local time (do NOT add one), do NOT edit tasks/${id}.md by hand, and this NEVER goes to the user.\n` + - `Need context from past runs? Read tasks/${id}.md first.]`; - const created = withInbound(session, (db) => { insertTaskRow(db, { id, seriesId: id, processAfter, recurrence, - content: JSON.stringify({ prompt: promptWithLog, script, originSessionId }), + content: JSON.stringify({ prompt, script, originSessionId }), }); return selectTask(db, id); }); @@ -309,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( @@ -329,22 +323,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 = formatLocalStamp(new Date(), TIMEZONE); - 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,9 +643,9 @@ registerResource({ 'append-log': { access: 'open', description: - 'Append a one-line run summary to a task run log (tasks/.md).\n\nThe host stamps the local 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: every task run auto-logs its final text, so use this only for additive mid-run notes. The host stamps the local 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: [ { @@ -674,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 d0ea0151730..cfa0a966085 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -213,11 +213,11 @@ 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, diff --git a/src/delivery.test.ts b/src/delivery.test.ts index 159d9708a87..7d0c6caaae5 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} \d{2}:\d{2} — 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 4ff8a935448..491aa48cb1b 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'; @@ -262,6 +269,24 @@ async function deliverMessage( return; } + // 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. + 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/modules/agent-to-agent/create-agent.ts b/src/modules/agent-to-agent/create-agent.ts index 4f8e557c8c5..16e22d17e0e 100644 --- a/src/modules/agent-to-agent/create-agent.ts +++ b/src/modules/agent-to-agent/create-agent.ts @@ -181,6 +181,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/recurrence.test.ts b/src/modules/scheduling/recurrence.test.ts index c470e5a2b73..7d2cc0584fb 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} \d{2}:\d{2} — /m); // appendRunLog's local-time stamp + }); }); diff --git a/src/modules/scheduling/recurrence.ts b/src/modules/scheduling/recurrence.ts index 833d4fdb1c9..99e68945448 100644 --- a/src/modules/scheduling/recurrence.ts +++ b/src/modules/scheduling/recurrence.ts @@ -11,18 +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 { formatLocalStamp } from '../../timezone.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 @@ -39,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 = formatLocalStamp(new Date(), TIMEZONE); - 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..49b935b60a6 --- /dev/null +++ b/src/modules/scheduling/run-log.ts @@ -0,0 +1,33 @@ +/** + * 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 run's final text produces + * (container/agent-runner poll-loop auto-append; delivery.ts routes it here) + */ +import fs from 'fs'; + +import { GROUPS_DIR, TIMEZONE } from '../../config.js'; +import { getAgentGroup } from '../../db/agent-groups.js'; +import { formatLocalStamp } from '../../timezone.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 = formatLocalStamp(new Date(), TIMEZONE); + 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 bfafe5991fc..7f624db1ba6 100644 --- a/src/session-manager.ts +++ b/src/session-manager.ts @@ -160,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