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.
This commit is contained in:
Akshan Krithick
GitHub
139
.claude/skills/add-compact/SKILL.md
Normal file
139
.claude/skills/add-compact/SKILL.md
Normal file
@@ -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: `@<assistant> /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: `@<assistant> /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 `@<assistant> /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 <your-fork> /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 @<assistant> /compact from non-main as non-admin, verify denial
|
||||
# Manual: send @<assistant> /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.
|
||||
214
.claude/skills/add-compact/add/src/session-commands.test.ts
Normal file
214
.claude/skills/add-compact/add/src/session-commands.test.ts
Normal file
@@ -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> = {}): NewMessage {
|
||||
return {
|
||||
id: 'msg-1',
|
||||
chat_jid: 'group@test',
|
||||
sender: 'user@test',
|
||||
sender_name: 'User',
|
||||
content,
|
||||
timestamp: '100',
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function makeDeps(overrides: Partial<SessionCommandDeps> = {}): 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('<formatted>'),
|
||||
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('<formatted>', 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'));
|
||||
});
|
||||
});
|
||||
143
.claude/skills/add-compact/add/src/session-commands.ts
Normal file
143
.claude/skills/add-compact/add/src/session-commands.ts
Normal file
@@ -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<void>;
|
||||
setTyping: (typing: boolean) => Promise<void>;
|
||||
runAgent: (
|
||||
prompt: string,
|
||||
onOutput: (result: AgentResult) => Promise<void>,
|
||||
) => 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(/<internal>[\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 };
|
||||
}
|
||||
16
.claude/skills/add-compact/manifest.yaml
Normal file
16
.claude/skills/add-compact/manifest.yaml
Normal file
@@ -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"
|
||||
@@ -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<string, string>;
|
||||
}
|
||||
|
||||
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<SDKUserMessage> {
|
||||
while (true) {
|
||||
while (this.queue.length > 0) {
|
||||
yield this.queue.shift()!;
|
||||
}
|
||||
if (this.done) return;
|
||||
await new Promise<void>(r => { this.waiting = r; });
|
||||
this.waiting = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function readStdin(): Promise<string> {
|
||||
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<string, unknown>),
|
||||
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<string | null> {
|
||||
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<string, string | undefined>,
|
||||
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<void> {
|
||||
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<string, string | undefined> = { ...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();
|
||||
@@ -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
|
||||
640
.claude/skills/add-compact/modify/src/index.ts
Normal file
640
.claude/skills/add-compact/modify/src/index.ts
Normal file
@@ -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<string, string> = {};
|
||||
let registeredGroups: Record<string, RegisteredGroup> = {};
|
||||
let lastAgentTimestamp: Record<string, string> = {};
|
||||
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<string, RegisteredGroup>,
|
||||
): 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<boolean> {
|
||||
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<typeof setTimeout> | 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 <internal>...</internal> blocks — agent uses these for internal reasoning
|
||||
const text = raw.replace(/<internal>[\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<void>,
|
||||
): 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<void> {
|
||||
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<string, NewMessage[]>();
|
||||
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<void> {
|
||||
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);
|
||||
});
|
||||
}
|
||||
25
.claude/skills/add-compact/modify/src/index.ts.intent.md
Normal file
25
.claude/skills/add-compact/modify/src/index.ts.intent.md
Normal file
@@ -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)
|
||||
188
.claude/skills/add-compact/tests/add-compact.test.ts
Normal file
188
.claude/skills/add-compact/tests/add-compact.test.ts
Normal file
@@ -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');
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user