From 8521e42f7b8d991d7e842c9a8d6915464ec5d0cb Mon Sep 17 00:00:00 2001 From: Akshan Krithick <97239696+akshan-main@users.noreply.github.com> Date: Sun, 8 Mar 2026 14:59:17 -0700 Subject: [PATCH] Add /compact skill for manual context compaction (#817) * feat: add /compact skill for manual context compaction added /compact session command to fight context rot in long-running sessions. Uses Claude Agent SDK's built-in /compact command with auth gating (main-group or is_from_me only). * simplify: remove group-queue modification, streamline denied path confirmed against fresh-clone merge. * refactor: extract handleSessionCommand from index.ts into session-commands.ts Verified: 345/345 tests pass on fresh-clone merge. --- .claude/skills/add-compact/SKILL.md | 139 ++++ .../add/src/session-commands.test.ts | 214 ++++++ .../add-compact/add/src/session-commands.ts | 143 ++++ .claude/skills/add-compact/manifest.yaml | 16 + .../container/agent-runner/src/index.ts | 688 ++++++++++++++++++ .../agent-runner/src/index.ts.intent.md | 29 + .../skills/add-compact/modify/src/index.ts | 640 ++++++++++++++++ .../add-compact/modify/src/index.ts.intent.md | 25 + .../add-compact/tests/add-compact.test.ts | 188 +++++ 9 files changed, 2082 insertions(+) create mode 100644 .claude/skills/add-compact/SKILL.md create mode 100644 .claude/skills/add-compact/add/src/session-commands.test.ts create mode 100644 .claude/skills/add-compact/add/src/session-commands.ts create mode 100644 .claude/skills/add-compact/manifest.yaml create mode 100644 .claude/skills/add-compact/modify/container/agent-runner/src/index.ts create mode 100644 .claude/skills/add-compact/modify/container/agent-runner/src/index.ts.intent.md create mode 100644 .claude/skills/add-compact/modify/src/index.ts create mode 100644 .claude/skills/add-compact/modify/src/index.ts.intent.md create mode 100644 .claude/skills/add-compact/tests/add-compact.test.ts diff --git a/.claude/skills/add-compact/SKILL.md b/.claude/skills/add-compact/SKILL.md new file mode 100644 index 0000000..1b75152 --- /dev/null +++ b/.claude/skills/add-compact/SKILL.md @@ -0,0 +1,139 @@ +--- +name: add-compact +description: Add /compact command for manual context compaction. Solves context rot in long sessions by forwarding the SDK's built-in /compact slash command. Main-group or trusted sender only. +--- + +# Add /compact Command + +Adds a `/compact` session command that compacts conversation history to fight context rot in long-running sessions. Uses the Claude Agent SDK's built-in `/compact` slash command — no synthetic system prompts. + +**Session contract:** `/compact` keeps the same logical session alive. The SDK returns a new session ID after compaction (via the `init` system message), which the agent-runner forwards to the orchestrator as `newSessionId`. No destructive reset occurs — the agent retains summarized context. + +## Phase 1: Pre-flight + +Read `.nanoclaw/state.yaml`. If `add-compact` is in `applied_skills`, skip to Phase 3 (Verify). + +## Phase 2: Apply Code Changes + +### Initialize skills system (if needed) + +If `.nanoclaw/` directory doesn't exist: + +```bash +npx tsx scripts/apply-skill.ts --init +``` + +### Apply the skill + +```bash +npx tsx scripts/apply-skill.ts .claude/skills/add-compact +``` + +This deterministically: +- Adds `src/session-commands.ts` (extract and authorize session commands) +- Adds `src/session-commands.test.ts` (unit tests for command parsing and auth) +- Three-way merges session command interception into `src/index.ts` (both `processGroupMessages` and `startMessageLoop`) +- Three-way merges slash command handling into `container/agent-runner/src/index.ts` +- Records application in `.nanoclaw/state.yaml` + +If merge conflicts occur, read the intent files: +- `modify/src/index.ts.intent.md` +- `modify/container/agent-runner/src/index.ts.intent.md` + +### Validate + +```bash +npm test +npm run build +``` + +### Rebuild container + +```bash +./container/build.sh +``` + +### Restart service + +```bash +launchctl kickstart -k gui/$(id -u)/com.nanoclaw # macOS +# Linux: systemctl --user restart nanoclaw +``` + +## Phase 3: Verify + +### Integration Test + +1. Start NanoClaw in dev mode: `npm run dev` +2. From the **main group** (self-chat), send exactly: `/compact` +3. Verify: + - The agent acknowledges compaction (e.g., "Conversation compacted.") + - The session continues — send a follow-up message and verify the agent responds coherently + - A conversation archive is written to `groups/{folder}/conversations/` (by the PreCompact hook) + - Container logs show `Compact boundary observed` (confirms SDK actually compacted) + - If `compact_boundary` was NOT observed, the response says "compact_boundary was not observed" +4. From a **non-main group** as a non-admin user, send: `@ /compact` +5. Verify: + - The bot responds with "Session commands require admin access." + - No compaction occurs, no container is spawned for the command +6. From a **non-main group** as the admin (device owner / `is_from_me`), send: `@ /compact` +7. Verify: + - Compaction proceeds normally (same behavior as main group) +8. While an **active container** is running for the main group, send `/compact` +9. Verify: + - The active container is signaled to close (authorized senders only — untrusted senders cannot kill in-flight work) + - Compaction proceeds via a new container once the active one exits + - The command is not dropped (no cursor race) +10. Send a normal message, then `/compact`, then another normal message in quick succession (same polling batch): +11. Verify: + - Pre-compact messages are sent to the agent first (check container logs for two `runAgent` calls) + - Compaction proceeds after pre-compact messages are processed + - Messages **after** `/compact` in the batch are preserved (cursor advances to `/compact`'s timestamp only) and processed on the next poll cycle +12. From a **non-main group** as a non-admin user, send `@ /compact`: +13. Verify: + - Denial message is sent ("Session commands require admin access.") + - The `/compact` is consumed (cursor advanced) — it does NOT replay on future polls + - Other messages in the same batch are also consumed (cursor is a high-water mark — this is an accepted tradeoff for the narrow edge case of denied `/compact` + other messages in the same polling interval) + - No container is killed or interrupted +14. From a **non-main group** (with `requiresTrigger` enabled) as a non-admin user, send bare `/compact` (no trigger prefix): +15. Verify: + - No denial message is sent (trigger policy prevents untrusted bot responses) + - The `/compact` is consumed silently + - Note: in groups where `requiresTrigger` is `false`, a denial message IS sent because the sender is considered reachable +16. After compaction, verify **no auto-compaction** behavior — only manual `/compact` triggers it + +### Validation on Fresh Clone + +```bash +git clone /tmp/nanoclaw-test +cd /tmp/nanoclaw-test +claude # then run /add-compact +npm run build +npm test +./container/build.sh +# Manual: send /compact from main group, verify compaction + continuation +# Manual: send @ /compact from non-main as non-admin, verify denial +# Manual: send @ /compact from non-main as admin, verify allowed +# Manual: verify no auto-compaction behavior +``` + +## Security Constraints + +- **Main-group or trusted/admin sender only.** The main group is the user's private self-chat and is trusted (see `docs/SECURITY.md`). Non-main groups are untrusted — a careless or malicious user could wipe the agent's short-term memory. However, the device owner (`is_from_me`) is always trusted and can compact from any group. +- **No auto-compaction.** This skill implements manual compaction only. Automatic threshold-based compaction is a separate concern and should be a separate skill. +- **No config file.** NanoClaw's philosophy is customization through code changes, not configuration sprawl. +- **Transcript archived before compaction.** The existing `PreCompact` hook in the agent-runner archives the full transcript to `conversations/` before the SDK compacts it. +- **Session continues after compaction.** This is not a destructive reset. The conversation continues with summarized context. + +## What This Does NOT Do + +- No automatic compaction threshold (add separately if desired) +- No `/clear` command (separate skill, separate semantics — `/clear` is a destructive reset) +- No cross-group compaction (each group's session is isolated) +- No changes to the container image, Dockerfile, or build script + +## Troubleshooting + +- **"Session commands require admin access"**: Only the device owner (`is_from_me`) or main-group senders can use `/compact`. Other users are denied. +- **No compact_boundary in logs**: The SDK may not emit this event in all versions. Check the agent-runner logs for the warning message. Compaction may still have succeeded. +- **Pre-compact failure**: If messages before `/compact` fail to process, the error message says "Failed to process messages before /compact." The cursor advances past sent output to prevent duplicates; `/compact` remains pending for the next attempt. diff --git a/.claude/skills/add-compact/add/src/session-commands.test.ts b/.claude/skills/add-compact/add/src/session-commands.test.ts new file mode 100644 index 0000000..7cbc680 --- /dev/null +++ b/.claude/skills/add-compact/add/src/session-commands.test.ts @@ -0,0 +1,214 @@ +import { describe, it, expect, vi } from 'vitest'; +import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js'; +import type { NewMessage } from './types.js'; +import type { SessionCommandDeps } from './session-commands.js'; + +describe('extractSessionCommand', () => { + const trigger = /^@Andy\b/i; + + it('detects bare /compact', () => { + expect(extractSessionCommand('/compact', trigger)).toBe('/compact'); + }); + + it('detects /compact with trigger prefix', () => { + expect(extractSessionCommand('@Andy /compact', trigger)).toBe('/compact'); + }); + + it('rejects /compact with extra text', () => { + expect(extractSessionCommand('/compact now please', trigger)).toBeNull(); + }); + + it('rejects partial matches', () => { + expect(extractSessionCommand('/compaction', trigger)).toBeNull(); + }); + + it('rejects regular messages', () => { + expect(extractSessionCommand('please compact the conversation', trigger)).toBeNull(); + }); + + it('handles whitespace', () => { + expect(extractSessionCommand(' /compact ', trigger)).toBe('/compact'); + }); + + it('is case-sensitive for the command', () => { + expect(extractSessionCommand('/Compact', trigger)).toBeNull(); + }); +}); + +describe('isSessionCommandAllowed', () => { + it('allows main group regardless of sender', () => { + expect(isSessionCommandAllowed(true, false)).toBe(true); + }); + + it('allows trusted/admin sender (is_from_me) in non-main group', () => { + expect(isSessionCommandAllowed(false, true)).toBe(true); + }); + + it('denies untrusted sender in non-main group', () => { + expect(isSessionCommandAllowed(false, false)).toBe(false); + }); + + it('allows trusted sender in main group', () => { + expect(isSessionCommandAllowed(true, true)).toBe(true); + }); +}); + +function makeMsg(content: string, overrides: Partial = {}): NewMessage { + return { + id: 'msg-1', + chat_jid: 'group@test', + sender: 'user@test', + sender_name: 'User', + content, + timestamp: '100', + ...overrides, + }; +} + +function makeDeps(overrides: Partial = {}): SessionCommandDeps { + return { + sendMessage: vi.fn().mockResolvedValue(undefined), + setTyping: vi.fn().mockResolvedValue(undefined), + runAgent: vi.fn().mockResolvedValue('success'), + closeStdin: vi.fn(), + advanceCursor: vi.fn(), + formatMessages: vi.fn().mockReturnValue(''), + canSenderInteract: vi.fn().mockReturnValue(true), + ...overrides, + }; +} + +const trigger = /^@Andy\b/i; + +describe('handleSessionCommand', () => { + it('returns handled:false when no session command found', async () => { + const deps = makeDeps(); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('hello')], + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result.handled).toBe(false); + }); + + it('handles authorized /compact in main group', async () => { + const deps = makeDeps(); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/compact')], + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function)); + expect(deps.advanceCursor).toHaveBeenCalledWith('100'); + }); + + it('sends denial to interactable sender in non-main group', async () => { + const deps = makeDeps(); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/compact', { is_from_me: false })], + isMainGroup: false, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.sendMessage).toHaveBeenCalledWith('Session commands require admin access.'); + expect(deps.runAgent).not.toHaveBeenCalled(); + expect(deps.advanceCursor).toHaveBeenCalledWith('100'); + }); + + it('silently consumes denied command when sender cannot interact', async () => { + const deps = makeDeps({ canSenderInteract: vi.fn().mockReturnValue(false) }); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/compact', { is_from_me: false })], + isMainGroup: false, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.sendMessage).not.toHaveBeenCalled(); + expect(deps.advanceCursor).toHaveBeenCalledWith('100'); + }); + + it('processes pre-compact messages before /compact', async () => { + const deps = makeDeps(); + const msgs = [ + makeMsg('summarize this', { timestamp: '99' }), + makeMsg('/compact', { timestamp: '100' }), + ]; + const result = await handleSessionCommand({ + missedMessages: msgs, + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.formatMessages).toHaveBeenCalledWith([msgs[0]], 'UTC'); + // Two runAgent calls: pre-compact + /compact + expect(deps.runAgent).toHaveBeenCalledTimes(2); + expect(deps.runAgent).toHaveBeenCalledWith('', expect.any(Function)); + expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function)); + }); + + it('allows is_from_me sender in non-main group', async () => { + const deps = makeDeps(); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/compact', { is_from_me: true })], + isMainGroup: false, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.runAgent).toHaveBeenCalledWith('/compact', expect.any(Function)); + }); + + it('reports failure when command-stage runAgent returns error without streamed status', async () => { + // runAgent resolves 'error' but callback never gets status: 'error' + const deps = makeDeps({ runAgent: vi.fn().mockImplementation(async (prompt, onOutput) => { + await onOutput({ status: 'success', result: null }); + return 'error'; + })}); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/compact')], + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.sendMessage).toHaveBeenCalledWith(expect.stringContaining('failed')); + }); + + it('returns success:false on pre-compact failure with no output', async () => { + const deps = makeDeps({ runAgent: vi.fn().mockResolvedValue('error') }); + const msgs = [ + makeMsg('summarize this', { timestamp: '99' }), + makeMsg('/compact', { timestamp: '100' }), + ]; + const result = await handleSessionCommand({ + missedMessages: msgs, + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: false }); + expect(deps.sendMessage).toHaveBeenCalledWith(expect.stringContaining('Failed to process')); + }); +}); diff --git a/.claude/skills/add-compact/add/src/session-commands.ts b/.claude/skills/add-compact/add/src/session-commands.ts new file mode 100644 index 0000000..69ea041 --- /dev/null +++ b/.claude/skills/add-compact/add/src/session-commands.ts @@ -0,0 +1,143 @@ +import type { NewMessage } from './types.js'; +import { logger } from './logger.js'; + +/** + * Extract a session slash command from a message, stripping the trigger prefix if present. + * Returns the slash command (e.g., '/compact') or null if not a session command. + */ +export function extractSessionCommand(content: string, triggerPattern: RegExp): string | null { + let text = content.trim(); + text = text.replace(triggerPattern, '').trim(); + if (text === '/compact') return '/compact'; + return null; +} + +/** + * Check if a session command sender is authorized. + * Allowed: main group (any sender), or trusted/admin sender (is_from_me) in any group. + */ +export function isSessionCommandAllowed(isMainGroup: boolean, isFromMe: boolean): boolean { + return isMainGroup || isFromMe; +} + +/** Minimal agent result interface — matches the subset of ContainerOutput used here. */ +export interface AgentResult { + status: 'success' | 'error'; + result?: string | object | null; +} + +/** Dependencies injected by the orchestrator. */ +export interface SessionCommandDeps { + sendMessage: (text: string) => Promise; + setTyping: (typing: boolean) => Promise; + runAgent: ( + prompt: string, + onOutput: (result: AgentResult) => Promise, + ) => Promise<'success' | 'error'>; + closeStdin: () => void; + advanceCursor: (timestamp: string) => void; + formatMessages: (msgs: NewMessage[], timezone: string) => string; + /** Whether the denied sender would normally be allowed to interact (for denial messages). */ + canSenderInteract: (msg: NewMessage) => boolean; +} + +function resultToText(result: string | object | null | undefined): string { + if (!result) return ''; + const raw = typeof result === 'string' ? result : JSON.stringify(result); + return raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); +} + +/** + * Handle session command interception in processGroupMessages. + * Scans messages for a session command, handles auth + execution. + * Returns { handled: true, success } if a command was found; { handled: false } otherwise. + * success=false means the caller should retry (cursor was not advanced). + */ +export async function handleSessionCommand(opts: { + missedMessages: NewMessage[]; + isMainGroup: boolean; + groupName: string; + triggerPattern: RegExp; + timezone: string; + deps: SessionCommandDeps; +}): Promise<{ handled: false } | { handled: true; success: boolean }> { + const { missedMessages, isMainGroup, groupName, triggerPattern, timezone, deps } = opts; + + const cmdMsg = missedMessages.find( + (m) => extractSessionCommand(m.content, triggerPattern) !== null, + ); + const command = cmdMsg ? extractSessionCommand(cmdMsg.content, triggerPattern) : null; + + if (!command || !cmdMsg) return { handled: false }; + + if (!isSessionCommandAllowed(isMainGroup, cmdMsg.is_from_me === true)) { + // DENIED: send denial if the sender would normally be allowed to interact, + // then silently consume the command by advancing the cursor past it. + // Trade-off: other messages in the same batch are also consumed (cursor is + // a high-water mark). Acceptable for this narrow edge case. + if (deps.canSenderInteract(cmdMsg)) { + await deps.sendMessage('Session commands require admin access.'); + } + deps.advanceCursor(cmdMsg.timestamp); + return { handled: true, success: true }; + } + + // AUTHORIZED: process pre-compact messages first, then run the command + logger.info({ group: groupName, command }, 'Session command'); + + const cmdIndex = missedMessages.indexOf(cmdMsg); + const preCompactMsgs = missedMessages.slice(0, cmdIndex); + + // Send pre-compact messages to the agent so they're in the session context. + if (preCompactMsgs.length > 0) { + const prePrompt = deps.formatMessages(preCompactMsgs, timezone); + let hadPreError = false; + let preOutputSent = false; + + const preResult = await deps.runAgent(prePrompt, async (result) => { + if (result.status === 'error') hadPreError = true; + const text = resultToText(result.result); + if (text) { + await deps.sendMessage(text); + preOutputSent = true; + } + // Close stdin on session-update marker — emitted after query completes, + // so all results (including multi-result runs) are already written. + if (result.status === 'success' && result.result === null) { + deps.closeStdin(); + } + }); + + if (preResult === 'error' || hadPreError) { + logger.warn({ group: groupName }, 'Pre-compact processing failed, aborting session command'); + await deps.sendMessage(`Failed to process messages before ${command}. Try again.`); + if (preOutputSent) { + // Output was already sent — don't retry or it will duplicate. + // Advance cursor past pre-compact messages, leave command pending. + deps.advanceCursor(preCompactMsgs[preCompactMsgs.length - 1].timestamp); + return { handled: true, success: true }; + } + return { handled: true, success: false }; + } + } + + // Forward the literal slash command as the prompt (no XML formatting) + await deps.setTyping(true); + + let hadCmdError = false; + const cmdOutput = await deps.runAgent(command, async (result) => { + if (result.status === 'error') hadCmdError = true; + const text = resultToText(result.result); + if (text) await deps.sendMessage(text); + }); + + // Advance cursor to the command — messages AFTER it remain pending for next poll. + deps.advanceCursor(cmdMsg.timestamp); + await deps.setTyping(false); + + if (cmdOutput === 'error' || hadCmdError) { + await deps.sendMessage(`${command} failed. The session is unchanged.`); + } + + return { handled: true, success: true }; +} diff --git a/.claude/skills/add-compact/manifest.yaml b/.claude/skills/add-compact/manifest.yaml new file mode 100644 index 0000000..3ac9b31 --- /dev/null +++ b/.claude/skills/add-compact/manifest.yaml @@ -0,0 +1,16 @@ +skill: add-compact +version: 1.0.0 +description: "Add /compact command for manual context compaction via Claude Agent SDK" +core_version: 1.2.10 +adds: + - src/session-commands.ts + - src/session-commands.test.ts +modifies: + - src/index.ts + - container/agent-runner/src/index.ts +structured: + npm_dependencies: {} + env_additions: [] +conflicts: [] +depends: [] +test: "npx vitest run --config vitest.skills.config.ts .claude/skills/add-compact/tests/add-compact.test.ts" diff --git a/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts b/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts new file mode 100644 index 0000000..a8f4c3b --- /dev/null +++ b/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts @@ -0,0 +1,688 @@ +/** + * NanoClaw Agent Runner + * Runs inside a container, receives config via stdin, outputs result to stdout + * + * Input protocol: + * Stdin: Full ContainerInput JSON (read until EOF, like before) + * IPC: Follow-up messages written as JSON files to /workspace/ipc/input/ + * Files: {type:"message", text:"..."}.json — polled and consumed + * Sentinel: /workspace/ipc/input/_close — signals session end + * + * Stdout protocol: + * Each result is wrapped in OUTPUT_START_MARKER / OUTPUT_END_MARKER pairs. + * Multiple results may be emitted (one per agent teams result). + * Final marker after loop ends signals completion. + */ + +import fs from 'fs'; +import path from 'path'; +import { query, HookCallback, PreCompactHookInput, PreToolUseHookInput } from '@anthropic-ai/claude-agent-sdk'; +import { fileURLToPath } from 'url'; + +interface ContainerInput { + prompt: string; + sessionId?: string; + groupFolder: string; + chatJid: string; + isMain: boolean; + isScheduledTask?: boolean; + assistantName?: string; + secrets?: Record; +} + +interface ContainerOutput { + status: 'success' | 'error'; + result: string | null; + newSessionId?: string; + error?: string; +} + +interface SessionEntry { + sessionId: string; + fullPath: string; + summary: string; + firstPrompt: string; +} + +interface SessionsIndex { + entries: SessionEntry[]; +} + +interface SDKUserMessage { + type: 'user'; + message: { role: 'user'; content: string }; + parent_tool_use_id: null; + session_id: string; +} + +const IPC_INPUT_DIR = '/workspace/ipc/input'; +const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, '_close'); +const IPC_POLL_MS = 500; + +/** + * Push-based async iterable for streaming user messages to the SDK. + * Keeps the iterable alive until end() is called, preventing isSingleUserTurn. + */ +class MessageStream { + private queue: SDKUserMessage[] = []; + private waiting: (() => void) | null = null; + private done = false; + + push(text: string): void { + this.queue.push({ + type: 'user', + message: { role: 'user', content: text }, + parent_tool_use_id: null, + session_id: '', + }); + this.waiting?.(); + } + + end(): void { + this.done = true; + this.waiting?.(); + } + + async *[Symbol.asyncIterator](): AsyncGenerator { + while (true) { + while (this.queue.length > 0) { + yield this.queue.shift()!; + } + if (this.done) return; + await new Promise(r => { this.waiting = r; }); + this.waiting = null; + } + } +} + +async function readStdin(): Promise { + return new Promise((resolve, reject) => { + let data = ''; + process.stdin.setEncoding('utf8'); + process.stdin.on('data', chunk => { data += chunk; }); + process.stdin.on('end', () => resolve(data)); + process.stdin.on('error', reject); + }); +} + +const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; +const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---'; + +function writeOutput(output: ContainerOutput): void { + console.log(OUTPUT_START_MARKER); + console.log(JSON.stringify(output)); + console.log(OUTPUT_END_MARKER); +} + +function log(message: string): void { + console.error(`[agent-runner] ${message}`); +} + +function getSessionSummary(sessionId: string, transcriptPath: string): string | null { + const projectDir = path.dirname(transcriptPath); + const indexPath = path.join(projectDir, 'sessions-index.json'); + + if (!fs.existsSync(indexPath)) { + log(`Sessions index not found at ${indexPath}`); + return null; + } + + try { + const index: SessionsIndex = JSON.parse(fs.readFileSync(indexPath, 'utf-8')); + const entry = index.entries.find(e => e.sessionId === sessionId); + if (entry?.summary) { + return entry.summary; + } + } catch (err) { + log(`Failed to read sessions index: ${err instanceof Error ? err.message : String(err)}`); + } + + return null; +} + +/** + * Archive the full transcript to conversations/ before compaction. + */ +function createPreCompactHook(assistantName?: string): HookCallback { + return async (input, _toolUseId, _context) => { + const preCompact = input as PreCompactHookInput; + const transcriptPath = preCompact.transcript_path; + const sessionId = preCompact.session_id; + + if (!transcriptPath || !fs.existsSync(transcriptPath)) { + log('No transcript found for archiving'); + return {}; + } + + try { + const content = fs.readFileSync(transcriptPath, 'utf-8'); + const messages = parseTranscript(content); + + if (messages.length === 0) { + log('No messages to archive'); + return {}; + } + + const summary = getSessionSummary(sessionId, transcriptPath); + const name = summary ? sanitizeFilename(summary) : generateFallbackName(); + + const conversationsDir = '/workspace/group/conversations'; + fs.mkdirSync(conversationsDir, { recursive: true }); + + const date = new Date().toISOString().split('T')[0]; + const filename = `${date}-${name}.md`; + const filePath = path.join(conversationsDir, filename); + + const markdown = formatTranscriptMarkdown(messages, summary, assistantName); + fs.writeFileSync(filePath, markdown); + + log(`Archived conversation to ${filePath}`); + } catch (err) { + log(`Failed to archive transcript: ${err instanceof Error ? err.message : String(err)}`); + } + + return {}; + }; +} + +// Secrets to strip from Bash tool subprocess environments. +// These are needed by claude-code for API auth but should never +// be visible to commands Kit runs. +const SECRET_ENV_VARS = ['ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN']; + +function createSanitizeBashHook(): HookCallback { + return async (input, _toolUseId, _context) => { + const preInput = input as PreToolUseHookInput; + const command = (preInput.tool_input as { command?: string })?.command; + if (!command) return {}; + + const unsetPrefix = `unset ${SECRET_ENV_VARS.join(' ')} 2>/dev/null; `; + return { + hookSpecificOutput: { + hookEventName: 'PreToolUse', + updatedInput: { + ...(preInput.tool_input as Record), + command: unsetPrefix + command, + }, + }, + }; + }; +} + +function sanitizeFilename(summary: string): string { + return summary + .toLowerCase() + .replace(/[^a-z0-9]+/g, '-') + .replace(/^-+|-+$/g, '') + .slice(0, 50); +} + +function generateFallbackName(): string { + const time = new Date(); + return `conversation-${time.getHours().toString().padStart(2, '0')}${time.getMinutes().toString().padStart(2, '0')}`; +} + +interface ParsedMessage { + role: 'user' | 'assistant'; + content: string; +} + +function parseTranscript(content: string): ParsedMessage[] { + const messages: ParsedMessage[] = []; + + for (const line of content.split('\n')) { + if (!line.trim()) continue; + try { + const entry = JSON.parse(line); + if (entry.type === 'user' && entry.message?.content) { + const text = typeof entry.message.content === 'string' + ? entry.message.content + : entry.message.content.map((c: { text?: string }) => c.text || '').join(''); + if (text) messages.push({ role: 'user', content: text }); + } else if (entry.type === 'assistant' && entry.message?.content) { + const textParts = entry.message.content + .filter((c: { type: string }) => c.type === 'text') + .map((c: { text: string }) => c.text); + const text = textParts.join(''); + if (text) messages.push({ role: 'assistant', content: text }); + } + } catch { + } + } + + return messages; +} + +function formatTranscriptMarkdown(messages: ParsedMessage[], title?: string | null, assistantName?: string): string { + const now = new Date(); + const formatDateTime = (d: Date) => d.toLocaleString('en-US', { + month: 'short', + day: 'numeric', + hour: 'numeric', + minute: '2-digit', + hour12: true + }); + + const lines: string[] = []; + lines.push(`# ${title || 'Conversation'}`); + lines.push(''); + lines.push(`Archived: ${formatDateTime(now)}`); + lines.push(''); + lines.push('---'); + lines.push(''); + + for (const msg of messages) { + const sender = msg.role === 'user' ? 'User' : (assistantName || 'Assistant'); + const content = msg.content.length > 2000 + ? msg.content.slice(0, 2000) + '...' + : msg.content; + lines.push(`**${sender}**: ${content}`); + lines.push(''); + } + + return lines.join('\n'); +} + +/** + * Check for _close sentinel. + */ +function shouldClose(): boolean { + if (fs.existsSync(IPC_INPUT_CLOSE_SENTINEL)) { + try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ } + return true; + } + return false; +} + +/** + * Drain all pending IPC input messages. + * Returns messages found, or empty array. + */ +function drainIpcInput(): string[] { + try { + fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); + const files = fs.readdirSync(IPC_INPUT_DIR) + .filter(f => f.endsWith('.json')) + .sort(); + + const messages: string[] = []; + for (const file of files) { + const filePath = path.join(IPC_INPUT_DIR, file); + try { + const data = JSON.parse(fs.readFileSync(filePath, 'utf-8')); + fs.unlinkSync(filePath); + if (data.type === 'message' && data.text) { + messages.push(data.text); + } + } catch (err) { + log(`Failed to process input file ${file}: ${err instanceof Error ? err.message : String(err)}`); + try { fs.unlinkSync(filePath); } catch { /* ignore */ } + } + } + return messages; + } catch (err) { + log(`IPC drain error: ${err instanceof Error ? err.message : String(err)}`); + return []; + } +} + +/** + * Wait for a new IPC message or _close sentinel. + * Returns the messages as a single string, or null if _close. + */ +function waitForIpcMessage(): Promise { + return new Promise((resolve) => { + const poll = () => { + if (shouldClose()) { + resolve(null); + return; + } + const messages = drainIpcInput(); + if (messages.length > 0) { + resolve(messages.join('\n')); + return; + } + setTimeout(poll, IPC_POLL_MS); + }; + poll(); + }); +} + +/** + * Run a single query and stream results via writeOutput. + * Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false, + * allowing agent teams subagents to run to completion. + * Also pipes IPC messages into the stream during the query. + */ +async function runQuery( + prompt: string, + sessionId: string | undefined, + mcpServerPath: string, + containerInput: ContainerInput, + sdkEnv: Record, + resumeAt?: string, +): Promise<{ newSessionId?: string; lastAssistantUuid?: string; closedDuringQuery: boolean }> { + const stream = new MessageStream(); + stream.push(prompt); + + // Poll IPC for follow-up messages and _close sentinel during the query + let ipcPolling = true; + let closedDuringQuery = false; + const pollIpcDuringQuery = () => { + if (!ipcPolling) return; + if (shouldClose()) { + log('Close sentinel detected during query, ending stream'); + closedDuringQuery = true; + stream.end(); + ipcPolling = false; + return; + } + const messages = drainIpcInput(); + for (const text of messages) { + log(`Piping IPC message into active query (${text.length} chars)`); + stream.push(text); + } + setTimeout(pollIpcDuringQuery, IPC_POLL_MS); + }; + setTimeout(pollIpcDuringQuery, IPC_POLL_MS); + + let newSessionId: string | undefined; + let lastAssistantUuid: string | undefined; + let messageCount = 0; + let resultCount = 0; + + // Load global CLAUDE.md as additional system context (shared across all groups) + const globalClaudeMdPath = '/workspace/global/CLAUDE.md'; + let globalClaudeMd: string | undefined; + if (!containerInput.isMain && fs.existsSync(globalClaudeMdPath)) { + globalClaudeMd = fs.readFileSync(globalClaudeMdPath, 'utf-8'); + } + + // Discover additional directories mounted at /workspace/extra/* + // These are passed to the SDK so their CLAUDE.md files are loaded automatically + const extraDirs: string[] = []; + const extraBase = '/workspace/extra'; + if (fs.existsSync(extraBase)) { + for (const entry of fs.readdirSync(extraBase)) { + const fullPath = path.join(extraBase, entry); + if (fs.statSync(fullPath).isDirectory()) { + extraDirs.push(fullPath); + } + } + } + if (extraDirs.length > 0) { + log(`Additional directories: ${extraDirs.join(', ')}`); + } + + for await (const message of query({ + prompt: stream, + options: { + cwd: '/workspace/group', + additionalDirectories: extraDirs.length > 0 ? extraDirs : undefined, + resume: sessionId, + resumeSessionAt: resumeAt, + systemPrompt: globalClaudeMd + ? { type: 'preset' as const, preset: 'claude_code' as const, append: globalClaudeMd } + : undefined, + allowedTools: [ + 'Bash', + 'Read', 'Write', 'Edit', 'Glob', 'Grep', + 'WebSearch', 'WebFetch', + 'Task', 'TaskOutput', 'TaskStop', + 'TeamCreate', 'TeamDelete', 'SendMessage', + 'TodoWrite', 'ToolSearch', 'Skill', + 'NotebookEdit', + 'mcp__nanoclaw__*' + ], + env: sdkEnv, + permissionMode: 'bypassPermissions', + allowDangerouslySkipPermissions: true, + settingSources: ['project', 'user'], + mcpServers: { + nanoclaw: { + command: 'node', + args: [mcpServerPath], + env: { + NANOCLAW_CHAT_JID: containerInput.chatJid, + NANOCLAW_GROUP_FOLDER: containerInput.groupFolder, + NANOCLAW_IS_MAIN: containerInput.isMain ? '1' : '0', + }, + }, + }, + hooks: { + PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], + PreToolUse: [{ matcher: 'Bash', hooks: [createSanitizeBashHook()] }], + }, + } + })) { + messageCount++; + const msgType = message.type === 'system' ? `system/${(message as { subtype?: string }).subtype}` : message.type; + log(`[msg #${messageCount}] type=${msgType}`); + + if (message.type === 'assistant' && 'uuid' in message) { + lastAssistantUuid = (message as { uuid: string }).uuid; + } + + if (message.type === 'system' && message.subtype === 'init') { + newSessionId = message.session_id; + log(`Session initialized: ${newSessionId}`); + } + + if (message.type === 'system' && (message as { subtype?: string }).subtype === 'task_notification') { + const tn = message as { task_id: string; status: string; summary: string }; + log(`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`); + } + + if (message.type === 'result') { + resultCount++; + const textResult = 'result' in message ? (message as { result?: string }).result : null; + log(`Result #${resultCount}: subtype=${message.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`); + writeOutput({ + status: 'success', + result: textResult || null, + newSessionId + }); + } + } + + ipcPolling = false; + log(`Query done. Messages: ${messageCount}, results: ${resultCount}, lastAssistantUuid: ${lastAssistantUuid || 'none'}, closedDuringQuery: ${closedDuringQuery}`); + return { newSessionId, lastAssistantUuid, closedDuringQuery }; +} + +async function main(): Promise { + let containerInput: ContainerInput; + + try { + const stdinData = await readStdin(); + containerInput = JSON.parse(stdinData); + // Delete the temp file the entrypoint wrote — it contains secrets + try { fs.unlinkSync('/tmp/input.json'); } catch { /* may not exist */ } + log(`Received input for group: ${containerInput.groupFolder}`); + } catch (err) { + writeOutput({ + status: 'error', + result: null, + error: `Failed to parse input: ${err instanceof Error ? err.message : String(err)}` + }); + process.exit(1); + } + + // Build SDK env: merge secrets into process.env for the SDK only. + // Secrets never touch process.env itself, so Bash subprocesses can't see them. + const sdkEnv: Record = { ...process.env }; + for (const [key, value] of Object.entries(containerInput.secrets || {})) { + sdkEnv[key] = value; + } + + const __dirname = path.dirname(fileURLToPath(import.meta.url)); + const mcpServerPath = path.join(__dirname, 'ipc-mcp-stdio.js'); + + let sessionId = containerInput.sessionId; + fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); + + // Clean up stale _close sentinel from previous container runs + try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ } + + // Build initial prompt (drain any pending IPC messages too) + let prompt = containerInput.prompt; + if (containerInput.isScheduledTask) { + prompt = `[SCHEDULED TASK - The following message was sent automatically and is not coming directly from the user or group.]\n\n${prompt}`; + } + const pending = drainIpcInput(); + if (pending.length > 0) { + log(`Draining ${pending.length} pending IPC messages into initial prompt`); + prompt += '\n' + pending.join('\n'); + } + + // --- Slash command handling --- + // Only known session slash commands are handled here. This prevents + // accidental interception of user prompts that happen to start with '/'. + const KNOWN_SESSION_COMMANDS = new Set(['/compact']); + const trimmedPrompt = prompt.trim(); + const isSessionSlashCommand = KNOWN_SESSION_COMMANDS.has(trimmedPrompt); + + if (isSessionSlashCommand) { + log(`Handling session command: ${trimmedPrompt}`); + let slashSessionId: string | undefined; + let compactBoundarySeen = false; + let hadError = false; + let resultEmitted = false; + + try { + for await (const message of query({ + prompt: trimmedPrompt, + options: { + cwd: '/workspace/group', + resume: sessionId, + systemPrompt: undefined, + allowedTools: [], + env: sdkEnv, + permissionMode: 'bypassPermissions' as const, + allowDangerouslySkipPermissions: true, + settingSources: ['project', 'user'] as const, + hooks: { + PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], + }, + }, + })) { + const msgType = message.type === 'system' + ? `system/${(message as { subtype?: string }).subtype}` + : message.type; + log(`[slash-cmd] type=${msgType}`); + + if (message.type === 'system' && message.subtype === 'init') { + slashSessionId = message.session_id; + log(`Session after slash command: ${slashSessionId}`); + } + + // Observe compact_boundary to confirm compaction completed + if (message.type === 'system' && (message as { subtype?: string }).subtype === 'compact_boundary') { + compactBoundarySeen = true; + log('Compact boundary observed — compaction completed'); + } + + if (message.type === 'result') { + const resultSubtype = (message as { subtype?: string }).subtype; + const textResult = 'result' in message ? (message as { result?: string }).result : null; + + if (resultSubtype?.startsWith('error')) { + hadError = true; + writeOutput({ + status: 'error', + result: null, + error: textResult || 'Session command failed.', + newSessionId: slashSessionId, + }); + } else { + writeOutput({ + status: 'success', + result: textResult || 'Conversation compacted.', + newSessionId: slashSessionId, + }); + } + resultEmitted = true; + } + } + } catch (err) { + hadError = true; + const errorMsg = err instanceof Error ? err.message : String(err); + log(`Slash command error: ${errorMsg}`); + writeOutput({ status: 'error', result: null, error: errorMsg }); + } + + log(`Slash command done. compactBoundarySeen=${compactBoundarySeen}, hadError=${hadError}`); + + // Warn if compact_boundary was never observed — compaction may not have occurred + if (!hadError && !compactBoundarySeen) { + log('WARNING: compact_boundary was not observed. Compaction may not have completed.'); + } + + // Only emit final session marker if no result was emitted yet and no error occurred + if (!resultEmitted && !hadError) { + writeOutput({ + status: 'success', + result: compactBoundarySeen + ? 'Conversation compacted.' + : 'Compaction requested but compact_boundary was not observed.', + newSessionId: slashSessionId, + }); + } else if (!hadError) { + // Emit session-only marker so host updates session tracking + writeOutput({ status: 'success', result: null, newSessionId: slashSessionId }); + } + return; + } + // --- End slash command handling --- + + // Query loop: run query → wait for IPC message → run new query → repeat + let resumeAt: string | undefined; + try { + while (true) { + log(`Starting query (session: ${sessionId || 'new'}, resumeAt: ${resumeAt || 'latest'})...`); + + const queryResult = await runQuery(prompt, sessionId, mcpServerPath, containerInput, sdkEnv, resumeAt); + if (queryResult.newSessionId) { + sessionId = queryResult.newSessionId; + } + if (queryResult.lastAssistantUuid) { + resumeAt = queryResult.lastAssistantUuid; + } + + // If _close was consumed during the query, exit immediately. + // Don't emit a session-update marker (it would reset the host's + // idle timer and cause a 30-min delay before the next _close). + if (queryResult.closedDuringQuery) { + log('Close sentinel consumed during query, exiting'); + break; + } + + // Emit session update so host can track it + writeOutput({ status: 'success', result: null, newSessionId: sessionId }); + + log('Query ended, waiting for next IPC message...'); + + // Wait for the next message or _close sentinel + const nextMessage = await waitForIpcMessage(); + if (nextMessage === null) { + log('Close sentinel received, exiting'); + break; + } + + log(`Got new message (${nextMessage.length} chars), starting new query`); + prompt = nextMessage; + } + } catch (err) { + const errorMessage = err instanceof Error ? err.message : String(err); + log(`Agent error: ${errorMessage}`); + writeOutput({ + status: 'error', + result: null, + newSessionId: sessionId, + error: errorMessage + }); + process.exit(1); + } +} + +main(); diff --git a/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts.intent.md b/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts.intent.md new file mode 100644 index 0000000..2538ca6 --- /dev/null +++ b/.claude/skills/add-compact/modify/container/agent-runner/src/index.ts.intent.md @@ -0,0 +1,29 @@ +# Intent: container/agent-runner/src/index.ts + +## What Changed +- Added `KNOWN_SESSION_COMMANDS` whitelist (`/compact`) +- Added slash command handling block in `main()` between prompt building and query loop +- Slash commands use `query()` with string prompt (not MessageStream), `allowedTools: []`, no mcpServers +- Tracks `compactBoundarySeen`, `hadError`, `resultEmitted` flags +- Observes `compact_boundary` system event to confirm compaction +- PreCompact hook still registered for transcript archival +- Error subtype checking: `resultSubtype?.startsWith('error')` emits `status: 'error'` +- Container exits after slash command completes (no IPC wait loop) + +## Key Sections +- **KNOWN_SESSION_COMMANDS** (before query loop): Set containing `/compact` +- **Slash command block** (after prompt building, before query loop): Detects session command, runs query with minimal options, handles result/error/boundary events +- **Existing query loop**: Unchanged + +## Invariants (must-keep) +- ContainerInput/ContainerOutput interfaces +- readStdin, writeOutput, log utilities +- OUTPUT_START_MARKER / OUTPUT_END_MARKER protocol +- MessageStream class with push/end/asyncIterator +- IPC polling (drainIpcInput, waitForIpcMessage, shouldClose) +- runQuery function with all existing logic +- createPreCompactHook for transcript archival +- createSanitizeBashHook for secret stripping +- parseTranscript, formatTranscriptMarkdown helpers +- main() stdin parsing, SDK env setup, query loop +- SECRET_ENV_VARS list diff --git a/.claude/skills/add-compact/modify/src/index.ts b/.claude/skills/add-compact/modify/src/index.ts new file mode 100644 index 0000000..d7df95c --- /dev/null +++ b/.claude/skills/add-compact/modify/src/index.ts @@ -0,0 +1,640 @@ +import fs from 'fs'; +import path from 'path'; + +import { + ASSISTANT_NAME, + IDLE_TIMEOUT, + POLL_INTERVAL, + TIMEZONE, + TRIGGER_PATTERN, +} from './config.js'; +import './channels/index.js'; +import { + getChannelFactory, + getRegisteredChannelNames, +} from './channels/registry.js'; +import { + ContainerOutput, + runContainerAgent, + writeGroupsSnapshot, + writeTasksSnapshot, +} from './container-runner.js'; +import { + cleanupOrphans, + ensureContainerRuntimeRunning, +} from './container-runtime.js'; +import { + getAllChats, + getAllRegisteredGroups, + getAllSessions, + getAllTasks, + getMessagesSince, + getNewMessages, + getRegisteredGroup, + getRouterState, + initDatabase, + setRegisteredGroup, + setRouterState, + setSession, + storeChatMetadata, + storeMessage, +} from './db.js'; +import { GroupQueue } from './group-queue.js'; +import { resolveGroupFolderPath } from './group-folder.js'; +import { startIpcWatcher } from './ipc.js'; +import { findChannel, formatMessages, formatOutbound } from './router.js'; +import { + isSenderAllowed, + isTriggerAllowed, + loadSenderAllowlist, + shouldDropMessage, +} from './sender-allowlist.js'; +import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js'; +import { startSchedulerLoop } from './task-scheduler.js'; +import { Channel, NewMessage, RegisteredGroup } from './types.js'; +import { logger } from './logger.js'; + +// Re-export for backwards compatibility during refactor +export { escapeXml, formatMessages } from './router.js'; + +let lastTimestamp = ''; +let sessions: Record = {}; +let registeredGroups: Record = {}; +let lastAgentTimestamp: Record = {}; +let messageLoopRunning = false; + +const channels: Channel[] = []; +const queue = new GroupQueue(); + +function loadState(): void { + lastTimestamp = getRouterState('last_timestamp') || ''; + const agentTs = getRouterState('last_agent_timestamp'); + try { + lastAgentTimestamp = agentTs ? JSON.parse(agentTs) : {}; + } catch { + logger.warn('Corrupted last_agent_timestamp in DB, resetting'); + lastAgentTimestamp = {}; + } + sessions = getAllSessions(); + registeredGroups = getAllRegisteredGroups(); + logger.info( + { groupCount: Object.keys(registeredGroups).length }, + 'State loaded', + ); +} + +function saveState(): void { + setRouterState('last_timestamp', lastTimestamp); + setRouterState('last_agent_timestamp', JSON.stringify(lastAgentTimestamp)); +} + +function registerGroup(jid: string, group: RegisteredGroup): void { + let groupDir: string; + try { + groupDir = resolveGroupFolderPath(group.folder); + } catch (err) { + logger.warn( + { jid, folder: group.folder, err }, + 'Rejecting group registration with invalid folder', + ); + return; + } + + registeredGroups[jid] = group; + setRegisteredGroup(jid, group); + + // Create group folder + fs.mkdirSync(path.join(groupDir, 'logs'), { recursive: true }); + + logger.info( + { jid, name: group.name, folder: group.folder }, + 'Group registered', + ); +} + +/** + * Get available groups list for the agent. + * Returns groups ordered by most recent activity. + */ +export function getAvailableGroups(): import('./container-runner.js').AvailableGroup[] { + const chats = getAllChats(); + const registeredJids = new Set(Object.keys(registeredGroups)); + + return chats + .filter((c) => c.jid !== '__group_sync__' && c.is_group) + .map((c) => ({ + jid: c.jid, + name: c.name, + lastActivity: c.last_message_time, + isRegistered: registeredJids.has(c.jid), + })); +} + +/** @internal - exported for testing */ +export function _setRegisteredGroups( + groups: Record, +): void { + registeredGroups = groups; +} + +/** + * Process all pending messages for a group. + * Called by the GroupQueue when it's this group's turn. + */ +async function processGroupMessages(chatJid: string): Promise { + const group = registeredGroups[chatJid]; + if (!group) return true; + + const channel = findChannel(channels, chatJid); + if (!channel) { + logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); + return true; + } + + const isMainGroup = group.isMain === true; + + const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; + const missedMessages = getMessagesSince( + chatJid, + sinceTimestamp, + ASSISTANT_NAME, + ); + + if (missedMessages.length === 0) return true; + + // --- Session command interception (before trigger check) --- + const cmdResult = await handleSessionCommand({ + missedMessages, + isMainGroup, + groupName: group.name, + triggerPattern: TRIGGER_PATTERN, + timezone: TIMEZONE, + deps: { + sendMessage: (text) => channel.sendMessage(chatJid, text), + setTyping: (typing) => channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), + runAgent: (prompt, onOutput) => runAgent(group, prompt, chatJid, onOutput), + closeStdin: () => queue.closeStdin(chatJid), + advanceCursor: (ts) => { lastAgentTimestamp[chatJid] = ts; saveState(); }, + formatMessages, + canSenderInteract: (msg) => { + const hasTrigger = TRIGGER_PATTERN.test(msg.content.trim()); + const reqTrigger = !isMainGroup && group.requiresTrigger !== false; + return isMainGroup || !reqTrigger || (hasTrigger && ( + msg.is_from_me || + isTriggerAllowed(chatJid, msg.sender, loadSenderAllowlist()) + )); + }, + }, + }); + if (cmdResult.handled) return cmdResult.success; + // --- End session command interception --- + + // For non-main groups, check if trigger is required and present + if (!isMainGroup && group.requiresTrigger !== false) { + const allowlistCfg = loadSenderAllowlist(); + const hasTrigger = missedMessages.some( + (m) => + TRIGGER_PATTERN.test(m.content.trim()) && + (m.is_from_me || isTriggerAllowed(chatJid, m.sender, allowlistCfg)), + ); + if (!hasTrigger) { + return true; + } + } + + const prompt = formatMessages(missedMessages, TIMEZONE); + + // Advance cursor so the piping path in startMessageLoop won't re-fetch + // these messages. Save the old cursor so we can roll back on error. + const previousCursor = lastAgentTimestamp[chatJid] || ''; + lastAgentTimestamp[chatJid] = + missedMessages[missedMessages.length - 1].timestamp; + saveState(); + + logger.info( + { group: group.name, messageCount: missedMessages.length }, + 'Processing messages', + ); + + // Track idle timer for closing stdin when agent is idle + let idleTimer: ReturnType | null = null; + + const resetIdleTimer = () => { + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => { + logger.debug( + { group: group.name }, + 'Idle timeout, closing container stdin', + ); + queue.closeStdin(chatJid); + }, IDLE_TIMEOUT); + }; + + await channel.setTyping?.(chatJid, true); + let hadError = false; + let outputSentToUser = false; + + const output = await runAgent(group, prompt, chatJid, async (result) => { + // Streaming output callback — called for each agent result + if (result.result) { + const raw = + typeof result.result === 'string' + ? result.result + : JSON.stringify(result.result); + // Strip ... blocks — agent uses these for internal reasoning + const text = raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); + logger.info({ group: group.name }, `Agent output: ${raw.slice(0, 200)}`); + if (text) { + await channel.sendMessage(chatJid, text); + outputSentToUser = true; + } + // Only reset idle timer on actual results, not session-update markers (result: null) + resetIdleTimer(); + } + + if (result.status === 'success') { + queue.notifyIdle(chatJid); + } + + if (result.status === 'error') { + hadError = true; + } + }); + + await channel.setTyping?.(chatJid, false); + if (idleTimer) clearTimeout(idleTimer); + + if (output === 'error' || hadError) { + // If we already sent output to the user, don't roll back the cursor — + // the user got their response and re-processing would send duplicates. + if (outputSentToUser) { + logger.warn( + { group: group.name }, + 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', + ); + return true; + } + // Roll back cursor so retries can re-process these messages + lastAgentTimestamp[chatJid] = previousCursor; + saveState(); + logger.warn( + { group: group.name }, + 'Agent error, rolled back message cursor for retry', + ); + return false; + } + + return true; +} + +async function runAgent( + group: RegisteredGroup, + prompt: string, + chatJid: string, + onOutput?: (output: ContainerOutput) => Promise, +): Promise<'success' | 'error'> { + const isMain = group.isMain === true; + const sessionId = sessions[group.folder]; + + // Update tasks snapshot for container to read (filtered by group) + const tasks = getAllTasks(); + writeTasksSnapshot( + group.folder, + isMain, + tasks.map((t) => ({ + id: t.id, + groupFolder: t.group_folder, + prompt: t.prompt, + schedule_type: t.schedule_type, + schedule_value: t.schedule_value, + status: t.status, + next_run: t.next_run, + })), + ); + + // Update available groups snapshot (main group only can see all groups) + const availableGroups = getAvailableGroups(); + writeGroupsSnapshot( + group.folder, + isMain, + availableGroups, + new Set(Object.keys(registeredGroups)), + ); + + // Wrap onOutput to track session ID from streamed results + const wrappedOnOutput = onOutput + ? async (output: ContainerOutput) => { + if (output.newSessionId) { + sessions[group.folder] = output.newSessionId; + setSession(group.folder, output.newSessionId); + } + await onOutput(output); + } + : undefined; + + try { + const output = await runContainerAgent( + group, + { + prompt, + sessionId, + groupFolder: group.folder, + chatJid, + isMain, + assistantName: ASSISTANT_NAME, + }, + (proc, containerName) => + queue.registerProcess(chatJid, proc, containerName, group.folder), + wrappedOnOutput, + ); + + if (output.newSessionId) { + sessions[group.folder] = output.newSessionId; + setSession(group.folder, output.newSessionId); + } + + if (output.status === 'error') { + logger.error( + { group: group.name, error: output.error }, + 'Container agent error', + ); + return 'error'; + } + + return 'success'; + } catch (err) { + logger.error({ group: group.name, err }, 'Agent error'); + return 'error'; + } +} + +async function startMessageLoop(): Promise { + if (messageLoopRunning) { + logger.debug('Message loop already running, skipping duplicate start'); + return; + } + messageLoopRunning = true; + + logger.info(`NanoClaw running (trigger: @${ASSISTANT_NAME})`); + + while (true) { + try { + const jids = Object.keys(registeredGroups); + const { messages, newTimestamp } = getNewMessages( + jids, + lastTimestamp, + ASSISTANT_NAME, + ); + + if (messages.length > 0) { + logger.info({ count: messages.length }, 'New messages'); + + // Advance the "seen" cursor for all messages immediately + lastTimestamp = newTimestamp; + saveState(); + + // Deduplicate by group + const messagesByGroup = new Map(); + for (const msg of messages) { + const existing = messagesByGroup.get(msg.chat_jid); + if (existing) { + existing.push(msg); + } else { + messagesByGroup.set(msg.chat_jid, [msg]); + } + } + + for (const [chatJid, groupMessages] of messagesByGroup) { + const group = registeredGroups[chatJid]; + if (!group) continue; + + const channel = findChannel(channels, chatJid); + if (!channel) { + logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); + continue; + } + + const isMainGroup = group.isMain === true; + + // --- Session command interception (message loop) --- + // Scan ALL messages in the batch for a session command. + const loopCmdMsg = groupMessages.find( + (m) => extractSessionCommand(m.content, TRIGGER_PATTERN) !== null, + ); + + if (loopCmdMsg) { + // Only close active container if the sender is authorized — otherwise an + // untrusted user could kill in-flight work by sending /compact (DoS). + // closeStdin no-ops internally when no container is active. + if (isSessionCommandAllowed(isMainGroup, loopCmdMsg.is_from_me === true)) { + queue.closeStdin(chatJid); + } + // Enqueue so processGroupMessages handles auth + cursor advancement. + // Don't pipe via IPC — slash commands need a fresh container with + // string prompt (not MessageStream) for SDK recognition. + queue.enqueueMessageCheck(chatJid); + continue; + } + // --- End session command interception --- + + const needsTrigger = !isMainGroup && group.requiresTrigger !== false; + + // For non-main groups, only act on trigger messages. + // Non-trigger messages accumulate in DB and get pulled as + // context when a trigger eventually arrives. + if (needsTrigger) { + const allowlistCfg = loadSenderAllowlist(); + const hasTrigger = groupMessages.some( + (m) => + TRIGGER_PATTERN.test(m.content.trim()) && + (m.is_from_me || + isTriggerAllowed(chatJid, m.sender, allowlistCfg)), + ); + if (!hasTrigger) continue; + } + + // Pull all messages since lastAgentTimestamp so non-trigger + // context that accumulated between triggers is included. + const allPending = getMessagesSince( + chatJid, + lastAgentTimestamp[chatJid] || '', + ASSISTANT_NAME, + ); + const messagesToSend = + allPending.length > 0 ? allPending : groupMessages; + const formatted = formatMessages(messagesToSend, TIMEZONE); + + if (queue.sendMessage(chatJid, formatted)) { + logger.debug( + { chatJid, count: messagesToSend.length }, + 'Piped messages to active container', + ); + lastAgentTimestamp[chatJid] = + messagesToSend[messagesToSend.length - 1].timestamp; + saveState(); + // Show typing indicator while the container processes the piped message + channel + .setTyping?.(chatJid, true) + ?.catch((err) => + logger.warn({ chatJid, err }, 'Failed to set typing indicator'), + ); + } else { + // No active container — enqueue for a new one + queue.enqueueMessageCheck(chatJid); + } + } + } + } catch (err) { + logger.error({ err }, 'Error in message loop'); + } + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL)); + } +} + +/** + * Startup recovery: check for unprocessed messages in registered groups. + * Handles crash between advancing lastTimestamp and processing messages. + */ +function recoverPendingMessages(): void { + for (const [chatJid, group] of Object.entries(registeredGroups)) { + const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; + const pending = getMessagesSince(chatJid, sinceTimestamp, ASSISTANT_NAME); + if (pending.length > 0) { + logger.info( + { group: group.name, pendingCount: pending.length }, + 'Recovery: found unprocessed messages', + ); + queue.enqueueMessageCheck(chatJid); + } + } +} + +function ensureContainerSystemRunning(): void { + ensureContainerRuntimeRunning(); + cleanupOrphans(); +} + +async function main(): Promise { + ensureContainerSystemRunning(); + initDatabase(); + logger.info('Database initialized'); + loadState(); + + // Graceful shutdown handlers + const shutdown = async (signal: string) => { + logger.info({ signal }, 'Shutdown signal received'); + await queue.shutdown(10000); + for (const ch of channels) await ch.disconnect(); + process.exit(0); + }; + process.on('SIGTERM', () => shutdown('SIGTERM')); + process.on('SIGINT', () => shutdown('SIGINT')); + + // Channel callbacks (shared by all channels) + const channelOpts = { + onMessage: (chatJid: string, msg: NewMessage) => { + // Sender allowlist drop mode: discard messages from denied senders before storing + if (!msg.is_from_me && !msg.is_bot_message && registeredGroups[chatJid]) { + const cfg = loadSenderAllowlist(); + if ( + shouldDropMessage(chatJid, cfg) && + !isSenderAllowed(chatJid, msg.sender, cfg) + ) { + if (cfg.logDenied) { + logger.debug( + { chatJid, sender: msg.sender }, + 'sender-allowlist: dropping message (drop mode)', + ); + } + return; + } + } + storeMessage(msg); + }, + onChatMetadata: ( + chatJid: string, + timestamp: string, + name?: string, + channel?: string, + isGroup?: boolean, + ) => storeChatMetadata(chatJid, timestamp, name, channel, isGroup), + registeredGroups: () => registeredGroups, + }; + + // Create and connect all registered channels. + // Each channel self-registers via the barrel import above. + // Factories return null when credentials are missing, so unconfigured channels are skipped. + for (const channelName of getRegisteredChannelNames()) { + const factory = getChannelFactory(channelName)!; + const channel = factory(channelOpts); + if (!channel) { + logger.warn( + { channel: channelName }, + 'Channel installed but credentials missing — skipping. Check .env or re-run the channel skill.', + ); + continue; + } + channels.push(channel); + await channel.connect(); + } + if (channels.length === 0) { + logger.fatal('No channels connected'); + process.exit(1); + } + + // Start subsystems (independently of connection handler) + startSchedulerLoop({ + registeredGroups: () => registeredGroups, + getSessions: () => sessions, + queue, + onProcess: (groupJid, proc, containerName, groupFolder) => + queue.registerProcess(groupJid, proc, containerName, groupFolder), + sendMessage: async (jid, rawText) => { + const channel = findChannel(channels, jid); + if (!channel) { + logger.warn({ jid }, 'No channel owns JID, cannot send message'); + return; + } + const text = formatOutbound(rawText); + if (text) await channel.sendMessage(jid, text); + }, + }); + startIpcWatcher({ + sendMessage: (jid, text) => { + const channel = findChannel(channels, jid); + if (!channel) throw new Error(`No channel for JID: ${jid}`); + return channel.sendMessage(jid, text); + }, + registeredGroups: () => registeredGroups, + registerGroup, + syncGroups: async (force: boolean) => { + await Promise.all( + channels + .filter((ch) => ch.syncGroups) + .map((ch) => ch.syncGroups!(force)), + ); + }, + getAvailableGroups, + writeGroupsSnapshot: (gf, im, ag, rj) => + writeGroupsSnapshot(gf, im, ag, rj), + }); + queue.setProcessMessagesFn(processGroupMessages); + recoverPendingMessages(); + startMessageLoop().catch((err) => { + logger.fatal({ err }, 'Message loop crashed unexpectedly'); + process.exit(1); + }); +} + +// Guard: only run when executed directly, not when imported by tests +const isDirectRun = + process.argv[1] && + new URL(import.meta.url).pathname === + new URL(`file://${process.argv[1]}`).pathname; + +if (isDirectRun) { + main().catch((err) => { + logger.error({ err }, 'Failed to start NanoClaw'); + process.exit(1); + }); +} diff --git a/.claude/skills/add-compact/modify/src/index.ts.intent.md b/.claude/skills/add-compact/modify/src/index.ts.intent.md new file mode 100644 index 0000000..0f915d7 --- /dev/null +++ b/.claude/skills/add-compact/modify/src/index.ts.intent.md @@ -0,0 +1,25 @@ +# Intent: src/index.ts + +## What Changed +- Added `import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js'` +- Added `handleSessionCommand()` call in `processGroupMessages()` between `missedMessages.length === 0` check and trigger check +- Added session command interception in `startMessageLoop()` between `isMainGroup` check and `needsTrigger` block + +## Key Sections +- **Imports** (top of file): extractSessionCommand, handleSessionCommand, isSessionCommandAllowed from session-commands +- **processGroupMessages**: Calls `handleSessionCommand()` with deps (sendMessage, runAgent, closeStdin, advanceCursor, formatMessages, canSenderInteract), returns early if handled +- **startMessageLoop**: Session command detection, auth-gated closeStdin (prevents DoS), enqueue for processGroupMessages + +## Invariants (must-keep) +- State management (lastTimestamp, sessions, registeredGroups, lastAgentTimestamp) +- loadState/saveState functions +- registerGroup function with folder validation +- getAvailableGroups function +- processGroupMessages trigger logic, cursor management, idle timer, error rollback with duplicate prevention +- runAgent task/group snapshot writes, session tracking, wrappedOnOutput +- startMessageLoop with dedup-by-group and piping logic +- recoverPendingMessages startup recovery +- main() with channel setup, scheduler, IPC watcher, queue +- ensureContainerSystemRunning using container-runtime abstraction +- Graceful shutdown with queue.shutdown +- Sender allowlist integration (drop mode, trigger check) diff --git a/.claude/skills/add-compact/tests/add-compact.test.ts b/.claude/skills/add-compact/tests/add-compact.test.ts new file mode 100644 index 0000000..396d57b --- /dev/null +++ b/.claude/skills/add-compact/tests/add-compact.test.ts @@ -0,0 +1,188 @@ +import { describe, it, expect, beforeAll } from 'vitest'; +import fs from 'fs'; +import path from 'path'; + +const SKILL_DIR = path.resolve(__dirname, '..'); + +describe('add-compact skill package', () => { + describe('manifest', () => { + let content: string; + + beforeAll(() => { + content = fs.readFileSync(path.join(SKILL_DIR, 'manifest.yaml'), 'utf-8'); + }); + + it('has a valid manifest.yaml', () => { + expect(fs.existsSync(path.join(SKILL_DIR, 'manifest.yaml'))).toBe(true); + expect(content).toContain('skill: add-compact'); + expect(content).toContain('version: 1.0.0'); + }); + + it('has no npm dependencies', () => { + expect(content).toContain('npm_dependencies: {}'); + }); + + it('has no env_additions', () => { + expect(content).toContain('env_additions: []'); + }); + + it('lists all add files', () => { + expect(content).toContain('src/session-commands.ts'); + expect(content).toContain('src/session-commands.test.ts'); + }); + + it('lists all modify files', () => { + expect(content).toContain('src/index.ts'); + expect(content).toContain('container/agent-runner/src/index.ts'); + }); + + it('has no dependencies', () => { + expect(content).toContain('depends: []'); + }); + }); + + describe('add/ files', () => { + it('includes src/session-commands.ts with required exports', () => { + const filePath = path.join(SKILL_DIR, 'add', 'src', 'session-commands.ts'); + expect(fs.existsSync(filePath)).toBe(true); + + const content = fs.readFileSync(filePath, 'utf-8'); + expect(content).toContain('export function extractSessionCommand'); + expect(content).toContain('export function isSessionCommandAllowed'); + expect(content).toContain('export async function handleSessionCommand'); + expect(content).toContain("'/compact'"); + }); + + it('includes src/session-commands.test.ts with test cases', () => { + const filePath = path.join(SKILL_DIR, 'add', 'src', 'session-commands.test.ts'); + expect(fs.existsSync(filePath)).toBe(true); + + const content = fs.readFileSync(filePath, 'utf-8'); + expect(content).toContain('extractSessionCommand'); + expect(content).toContain('isSessionCommandAllowed'); + expect(content).toContain('detects bare /compact'); + expect(content).toContain('denies untrusted sender'); + }); + }); + + describe('modify/ files exist', () => { + const modifyFiles = [ + 'src/index.ts', + 'container/agent-runner/src/index.ts', + ]; + + for (const file of modifyFiles) { + it(`includes modify/${file}`, () => { + const filePath = path.join(SKILL_DIR, 'modify', file); + expect(fs.existsSync(filePath)).toBe(true); + }); + } + }); + + describe('intent files exist', () => { + const intentFiles = [ + 'src/index.ts.intent.md', + 'container/agent-runner/src/index.ts.intent.md', + ]; + + for (const file of intentFiles) { + it(`includes modify/${file}`, () => { + const filePath = path.join(SKILL_DIR, 'modify', file); + expect(fs.existsSync(filePath)).toBe(true); + }); + } + }); + + describe('modify/src/index.ts', () => { + let content: string; + + beforeAll(() => { + content = fs.readFileSync( + path.join(SKILL_DIR, 'modify', 'src', 'index.ts'), + 'utf-8', + ); + }); + + it('imports session command helpers', () => { + expect(content).toContain("import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed } from './session-commands.js'"); + }); + + it('uses const for missedMessages', () => { + expect(content).toMatch(/const missedMessages = getMessagesSince/); + }); + + it('delegates to handleSessionCommand in processGroupMessages', () => { + expect(content).toContain('Session command interception (before trigger check)'); + expect(content).toContain('handleSessionCommand('); + expect(content).toContain('cmdResult.handled'); + expect(content).toContain('cmdResult.success'); + }); + + it('passes deps to handleSessionCommand', () => { + expect(content).toContain('sendMessage:'); + expect(content).toContain('setTyping:'); + expect(content).toContain('runAgent:'); + expect(content).toContain('closeStdin:'); + expect(content).toContain('advanceCursor:'); + expect(content).toContain('formatMessages'); + expect(content).toContain('canSenderInteract:'); + }); + + it('has session command interception in startMessageLoop', () => { + expect(content).toContain('Session command interception (message loop)'); + expect(content).toContain('queue.enqueueMessageCheck(chatJid)'); + }); + + it('preserves core index.ts structure', () => { + expect(content).toContain('processGroupMessages'); + expect(content).toContain('startMessageLoop'); + expect(content).toContain('async function main()'); + expect(content).toContain('recoverPendingMessages'); + expect(content).toContain('ensureContainerSystemRunning'); + }); + }); + + describe('modify/container/agent-runner/src/index.ts', () => { + let content: string; + + beforeAll(() => { + content = fs.readFileSync( + path.join(SKILL_DIR, 'modify', 'container', 'agent-runner', 'src', 'index.ts'), + 'utf-8', + ); + }); + + it('defines KNOWN_SESSION_COMMANDS whitelist', () => { + expect(content).toContain("KNOWN_SESSION_COMMANDS"); + expect(content).toContain("'/compact'"); + }); + + it('uses query() with string prompt for slash commands', () => { + expect(content).toContain('prompt: trimmedPrompt'); + expect(content).toContain('allowedTools: []'); + }); + + it('observes compact_boundary system event', () => { + expect(content).toContain('compactBoundarySeen'); + expect(content).toContain("'compact_boundary'"); + expect(content).toContain('Compact boundary observed'); + }); + + it('handles error subtypes', () => { + expect(content).toContain("resultSubtype?.startsWith('error')"); + }); + + it('registers PreCompact hook for slash commands', () => { + expect(content).toContain('createPreCompactHook(containerInput.assistantName)'); + }); + + it('preserves core agent-runner structure', () => { + expect(content).toContain('async function runQuery'); + expect(content).toContain('class MessageStream'); + expect(content).toContain('function writeOutput'); + expect(content).toContain('function createPreCompactHook'); + expect(content).toContain('function createSanitizeBashHook'); + expect(content).toContain('async function main'); + }); + }); +});