From e7f49d77daccdb960e461a8e3e3c06529bd979f3 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Sat, 28 Mar 2026 21:14:25 +0900 Subject: [PATCH] feat: add paired review workspace flow --- runners/agent-runner/src/index.ts | 76 ++- runners/agent-runner/src/reviewer-runtime.ts | 220 +++++++++ .../test/reviewer-runtime.test.ts | 67 +++ runners/codex-runner/src/index.ts | 10 +- runners/codex-runner/src/reviewer-runtime.ts | 105 +++++ .../test/reviewer-runtime.test.ts | 54 +++ src/db.test.ts | 165 +++++++ src/db.ts | 441 ++++++++++++++++++ src/group-folder.ts | 29 ++ src/message-agent-executor.test.ts | 182 ++++++++ src/message-agent-executor.ts | 101 +++- src/message-runtime.ts | 28 +- src/paired-execution-context.test.ts | 244 ++++++++++ src/paired-execution-context.ts | 258 ++++++++++ src/paired-workspace-manager.test.ts | 274 +++++++++++ src/paired-workspace-manager.ts | 337 +++++++++++++ src/session-commands.test.ts | 32 ++ src/session-commands.ts | 14 + src/types.ts | 102 ++++ 19 files changed, 2719 insertions(+), 20 deletions(-) create mode 100644 runners/agent-runner/src/reviewer-runtime.ts create mode 100644 runners/agent-runner/test/reviewer-runtime.test.ts create mode 100644 runners/codex-runner/src/reviewer-runtime.ts create mode 100644 runners/codex-runner/test/reviewer-runtime.test.ts create mode 100644 src/paired-execution-context.test.ts create mode 100644 src/paired-execution-context.ts create mode 100644 src/paired-workspace-manager.test.ts create mode 100644 src/paired-workspace-manager.ts diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index efc2416..b6b8287 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -25,6 +25,11 @@ import { prependRoomRoleHeader, type RoomRoleContext, } from './room-role-context.js'; +import { + buildReviewerGitGuardEnv, + isReviewerMutatingShellCommand, + isReviewerRuntime, +} from './reviewer-runtime.js'; interface ContainerInput { prompt: string; @@ -466,6 +471,25 @@ function createSanitizeBashHook(): HookCallback { }; } +function createReviewerBashGuardHook(): HookCallback { + return async (input, _toolUseId, _context) => { + const preInput = input as PreToolUseHookInput; + const command = (preInput.tool_input as { command?: string })?.command; + if (!command || !isReviewerMutatingShellCommand(command)) { + return {}; + } + + return { + hookSpecificOutput: { + hookEventName: 'PreToolUse', + permissionDecision: 'deny', + permissionDecisionReason: + 'EJClaw reviewer runtime blocks mutating shell commands in paired review mode.', + }, + }; + }; +} + function sanitizeFilename(summary: string): string { return summary .toLowerCase() @@ -595,6 +619,7 @@ async function runQuery( mcpServerPath: string, containerInput: ContainerInput, sdkEnv: Record, + reviewerRuntime: boolean, ): Promise<{ newSessionId?: string; closedDuringQuery: boolean; @@ -658,6 +683,28 @@ async function runQuery( if (model) log(`Using model: ${model}`); if (thinking) log(`Thinking config: ${JSON.stringify(thinking)}`); if (effort) log(`Effort: ${effort}`); + if (reviewerRuntime) log('Reviewer runtime restrictions enabled'); + + const allowedTools = reviewerRuntime + ? [ + 'Bash', + 'Read', 'Glob', 'Grep', + 'WebSearch', 'WebFetch', + 'Task', 'TaskOutput', 'TaskStop', + 'TeamCreate', 'TeamDelete', 'SendMessage', + 'TodoWrite', 'ToolSearch', 'Skill', + 'mcp__ejclaw__*', + ] + : [ + 'Bash', + 'Read', 'Write', 'Edit', 'Glob', 'Grep', + 'WebSearch', 'WebFetch', + 'Task', 'TaskOutput', 'TaskStop', + 'TeamCreate', 'TeamDelete', 'SendMessage', + 'TodoWrite', 'ToolSearch', 'Skill', + 'NotebookEdit', + 'mcp__ejclaw__*' + ]; for await (const message of query({ prompt: stream, @@ -668,16 +715,7 @@ async function runQuery( effort, additionalDirectories: extraDirs.length > 0 ? extraDirs : undefined, resume: sessionId, - allowedTools: [ - 'Bash', - 'Read', 'Write', 'Edit', 'Glob', 'Grep', - 'WebSearch', 'WebFetch', - 'Task', 'TaskOutput', 'TaskStop', - 'TeamCreate', 'TeamDelete', 'SendMessage', - 'TodoWrite', 'ToolSearch', 'Skill', - 'NotebookEdit', - 'mcp__ejclaw__*' - ], + allowedTools, env: sdkEnv, permissionMode: 'bypassPermissions', allowDangerouslySkipPermissions: true, @@ -715,7 +753,14 @@ async function runQuery( }, hooks: { PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], - PreToolUse: [{ matcher: 'Bash', hooks: [createSanitizeBashHook()] }], + PreToolUse: [ + { + matcher: 'Bash', + hooks: reviewerRuntime + ? [createReviewerBashGuardHook(), createSanitizeBashHook()] + : [createSanitizeBashHook()], + }, + ], }, agentProgressSummaries: true, } @@ -954,6 +999,10 @@ async function main(): Promise { for (const [key, value] of Object.entries(containerInput.secrets || {})) { sdkEnv[key] = value; } + const reviewerRuntime = + process.env.EJCLAW_REVIEWER_RUNTIME === '1' || + isReviewerRuntime(containerInput.roomRoleContext); + const guardedSdkEnv = buildReviewerGitGuardEnv(sdkEnv, reviewerRuntime); const __dirname = path.dirname(fileURLToPath(import.meta.url)); const mcpServerPath = path.join(__dirname, 'ipc-mcp-stdio.js'); @@ -1001,7 +1050,7 @@ async function main(): Promise { resume: sessionId, systemPrompt: undefined, allowedTools: [], - env: sdkEnv, + env: guardedSdkEnv, permissionMode: 'bypassPermissions' as const, allowDangerouslySkipPermissions: true, settingSources: ['project', 'user'] as const, @@ -1088,7 +1137,8 @@ async function main(): Promise { sessionId, mcpServerPath, containerInput, - sdkEnv, + guardedSdkEnv, + reviewerRuntime, ); if (queryResult.newSessionId) { sessionId = queryResult.newSessionId; diff --git a/runners/agent-runner/src/reviewer-runtime.ts b/runners/agent-runner/src/reviewer-runtime.ts new file mode 100644 index 0000000..48c9fe3 --- /dev/null +++ b/runners/agent-runner/src/reviewer-runtime.ts @@ -0,0 +1,220 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import type { RoomRoleContext } from './room-role-context.js'; + +const BLOCKED_GIT_SUBCOMMANDS = new Set([ + 'add', + 'am', + 'apply', + 'branch', + 'checkout', + 'cherry-pick', + 'clean', + 'commit', + 'merge', + 'push', + 'rebase', + 'reset', + 'restore', + 'stash', + 'switch', + 'tag', + 'worktree', +]); + +const MUTATING_SHELL_PATTERNS = [ + /\bsed\s+-i\b/i, + /\bperl\s+-i\b/i, + /(^|[;&|])\s*(cat|echo|printf)\b[^#\n]*>>?/i, +]; + +export function isReviewerRuntime( + roomRoleContext?: RoomRoleContext, +): boolean { + return roomRoleContext?.role === 'reviewer'; +} + +function findGitSubcommand(args: string[]): string { + let skipNext = false; + for (const arg of args) { + if (skipNext) { + skipNext = false; + continue; + } + switch (arg) { + case '-c': + case '-C': + case '--git-dir': + case '--work-tree': + case '--namespace': + case '--exec-path': + case '--config-env': + skipNext = true; + continue; + default: + break; + } + if ( + arg.startsWith('-c') || + arg.startsWith('-C') || + arg.startsWith('--git-dir=') || + arg.startsWith('--work-tree=') || + arg.startsWith('--namespace=') || + arg.startsWith('--exec-path=') || + arg.startsWith('--config-env=') + ) { + continue; + } + if (arg.startsWith('-')) { + continue; + } + return arg; + } + return ''; +} + +function tokenizeShellWords(segment: string): string[] { + const tokens: string[] = []; + let current = ''; + let quote: '"' | "'" | null = null; + let escaped = false; + + for (const char of segment) { + if (escaped) { + current += char; + escaped = false; + continue; + } + if (char === '\\' && quote !== "'") { + escaped = true; + continue; + } + if (quote) { + if (char === quote) { + quote = null; + } else { + current += char; + } + continue; + } + if (char === '"' || char === "'") { + quote = char; + continue; + } + if (/\s/.test(char)) { + if (current) { + tokens.push(current); + current = ''; + } + continue; + } + current += char; + } + + if (current) { + tokens.push(current); + } + return tokens; +} + +function isBlockedGitCommand(command: string): boolean { + const segments = command + .split(/&&|\|\||[;\n]/) + .map((segment) => segment.trim()) + .filter(Boolean); + for (const segment of segments) { + const tokens = tokenizeShellWords(segment); + if (tokens[0] !== 'git') { + continue; + } + const subcommand = findGitSubcommand(tokens.slice(1)); + if (BLOCKED_GIT_SUBCOMMANDS.has(subcommand)) { + return true; + } + } + return false; +} + +function resolveGitBinary(baseEnv: NodeJS.ProcessEnv): string { + return execFileSync('bash', ['-lc', 'command -v git'], { + encoding: 'utf-8', + env: baseEnv, + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +export function buildReviewerGitGuardEnv( + baseEnv: NodeJS.ProcessEnv, + reviewerRuntime: boolean, +): NodeJS.ProcessEnv { + if (!reviewerRuntime) { + return baseEnv; + } + + const realGitPath = resolveGitBinary(baseEnv); + const wrapperDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-reviewer-git-'), + ); + const wrapperPath = path.join(wrapperDir, 'git'); + const blocked = [...BLOCKED_GIT_SUBCOMMANDS] + .map((value) => `'${value}'`) + .join(' '); + + const script = `#!/usr/bin/env bash +set -euo pipefail +real_git=${JSON.stringify(realGitPath)} +blocked_subcommands=(${blocked}) +subcmd="" +skip_next=0 +for arg in "$@"; do + if [[ "$skip_next" == "1" ]]; then + skip_next=0 + continue + fi + case "$arg" in + -c|-C|--git-dir|--work-tree|--namespace|--exec-path|--config-env) + skip_next=1 + continue + ;; + -c*|-C*|--git-dir=*|--work-tree=*|--namespace=*|--exec-path=*|--config-env=*) + continue + ;; + --*) + continue + ;; + -*) + continue + ;; + *) + subcmd="$arg" + break + ;; + esac +done +for blocked in "\${blocked_subcommands[@]}"; do + if [[ "$subcmd" == "$blocked" ]]; then + echo "EJClaw reviewer runtime blocks mutating git subcommands: $subcmd" >&2 + exit 1 + fi +done +exec "$real_git" "$@" +`; + + fs.writeFileSync(wrapperPath, script, { mode: 0o755 }); + return { + ...baseEnv, + EJCLAW_REAL_GIT: realGitPath, + PATH: `${wrapperDir}:${baseEnv.PATH || ''}`, + }; +} + +export function isReviewerMutatingShellCommand(command: string): boolean { + const normalized = command.trim(); + return ( + isBlockedGitCommand(normalized) || + MUTATING_SHELL_PATTERNS.some((pattern) => pattern.test(normalized)) + ); +} diff --git a/runners/agent-runner/test/reviewer-runtime.test.ts b/runners/agent-runner/test/reviewer-runtime.test.ts new file mode 100644 index 0000000..beef64c --- /dev/null +++ b/runners/agent-runner/test/reviewer-runtime.test.ts @@ -0,0 +1,67 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { describe, expect, it } from 'vitest'; + +import { + buildReviewerGitGuardEnv, + isReviewerMutatingShellCommand, + isReviewerRuntime, +} from '../src/reviewer-runtime.js'; + +describe('claude reviewer runtime guard', () => { + it('detects reviewer room metadata', () => { + expect( + isReviewerRuntime({ + serviceId: 'codex-review', + role: 'reviewer', + ownerServiceId: 'codex-main', + reviewerServiceId: 'codex-review', + failoverOwner: false, + }), + ).toBe(true); + }); + + it('flags mutating shell commands', () => { + expect(isReviewerMutatingShellCommand('git commit -m "x"')).toBe(true); + expect( + isReviewerMutatingShellCommand('git -c color.ui=false commit -m "x"'), + ).toBe(true); + expect(isReviewerMutatingShellCommand('sed -i s/a/b/ file.ts')).toBe( + true, + ); + expect(isReviewerMutatingShellCommand('git status')).toBe(false); + expect(isReviewerMutatingShellCommand('npm test')).toBe(false); + }); + + it('prepends a git wrapper to PATH for reviewer runtimes', () => { + const env = buildReviewerGitGuardEnv({ PATH: process.env.PATH }, true); + expect(env.PATH).toContain('ejclaw-reviewer-git-'); + expect(env.EJCLAW_REAL_GIT).toBeTruthy(); + }); + + it('blocks mutating git subcommands even when git options come first', () => { + const env = buildReviewerGitGuardEnv({ PATH: process.env.PATH }, true); + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-reviewer-test-')); + + try { + execFileSync('git', ['-c', 'color.ui=false', 'commit', '-m', 'x'], { + cwd, + env, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }); + throw new Error('expected git wrapper to block commit'); + } catch (error) { + const stderr = + error instanceof Error && 'stderr' in error + ? String((error as Error & { stderr?: string | Buffer }).stderr ?? '') + : ''; + expect(stderr).toContain( + 'EJClaw reviewer runtime blocks mutating git subcommands: commit', + ); + } + }); +}); diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index ee7f1f6..4fa3e5d 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -23,6 +23,10 @@ import { prependRoomRoleHeader, type RoomRoleContext, } from './room-role-context.js'; +import { + buildReviewerGitGuardEnv, + isReviewerRuntime, +} from './reviewer-runtime.js'; // ── Types ────────────────────────────────────────────────────────── @@ -358,9 +362,13 @@ async function runAppServerSession( containerInput: ContainerInput, prompt: string, ): Promise { + const reviewerRuntime = + process.env.EJCLAW_REVIEWER_RUNTIME === '1' || + isReviewerRuntime(containerInput.roomRoleContext); + const clientEnv = buildReviewerGitGuardEnv(process.env, reviewerRuntime); const client = new CodexAppServerClient({ cwd: EFFECTIVE_CWD, - env: process.env, + env: clientEnv, log, }); diff --git a/runners/codex-runner/src/reviewer-runtime.ts b/runners/codex-runner/src/reviewer-runtime.ts new file mode 100644 index 0000000..5f98e31 --- /dev/null +++ b/runners/codex-runner/src/reviewer-runtime.ts @@ -0,0 +1,105 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import type { RoomRoleContext } from './room-role-context.js'; + +const BLOCKED_GIT_SUBCOMMANDS = new Set([ + 'add', + 'am', + 'apply', + 'branch', + 'checkout', + 'cherry-pick', + 'clean', + 'commit', + 'merge', + 'push', + 'rebase', + 'reset', + 'restore', + 'stash', + 'switch', + 'tag', + 'worktree', +]); + +export function isReviewerRuntime( + roomRoleContext?: RoomRoleContext, +): boolean { + return roomRoleContext?.role === 'reviewer'; +} + +function resolveGitBinary(baseEnv: NodeJS.ProcessEnv): string { + return execFileSync('bash', ['-lc', 'command -v git'], { + encoding: 'utf-8', + env: baseEnv, + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +export function buildReviewerGitGuardEnv( + baseEnv: NodeJS.ProcessEnv, + reviewerRuntime: boolean, +): NodeJS.ProcessEnv { + if (!reviewerRuntime) { + return baseEnv; + } + + const realGitPath = resolveGitBinary(baseEnv); + const wrapperDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-reviewer-git-'), + ); + const wrapperPath = path.join(wrapperDir, 'git'); + const blocked = [...BLOCKED_GIT_SUBCOMMANDS] + .map((value) => `'${value}'`) + .join(' '); + + const script = `#!/usr/bin/env bash +set -euo pipefail +real_git=${JSON.stringify(realGitPath)} +blocked_subcommands=(${blocked}) +subcmd="" +skip_next=0 +for arg in "$@"; do + if [[ "$skip_next" == "1" ]]; then + skip_next=0 + continue + fi + case "$arg" in + -c|-C|--git-dir|--work-tree|--namespace|--exec-path|--config-env) + skip_next=1 + continue + ;; + -c*|-C*|--git-dir=*|--work-tree=*|--namespace=*|--exec-path=*|--config-env=*) + continue + ;; + --*) + continue + ;; + -*) + continue + ;; + *) + subcmd="$arg" + break + ;; + esac +done +for blocked in "\${blocked_subcommands[@]}"; do + if [[ "$subcmd" == "$blocked" ]]; then + echo "EJClaw reviewer runtime blocks mutating git subcommands: $subcmd" >&2 + exit 1 + fi +done +exec "$real_git" "$@" +`; + + fs.writeFileSync(wrapperPath, script, { mode: 0o755 }); + return { + ...baseEnv, + EJCLAW_REAL_GIT: realGitPath, + PATH: `${wrapperDir}:${baseEnv.PATH || ''}`, + }; +} diff --git a/runners/codex-runner/test/reviewer-runtime.test.ts b/runners/codex-runner/test/reviewer-runtime.test.ts new file mode 100644 index 0000000..43e751c --- /dev/null +++ b/runners/codex-runner/test/reviewer-runtime.test.ts @@ -0,0 +1,54 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { describe, expect, it } from 'vitest'; + +import { + buildReviewerGitGuardEnv, + isReviewerRuntime, +} from '../src/reviewer-runtime.js'; + +describe('codex reviewer runtime guard', () => { + it('detects reviewer room metadata', () => { + expect( + isReviewerRuntime({ + serviceId: 'codex-review', + role: 'reviewer', + ownerServiceId: 'codex-main', + reviewerServiceId: 'codex-review', + failoverOwner: false, + }), + ).toBe(true); + }); + + it('prepends a git wrapper to PATH for reviewer runtimes', () => { + const env = buildReviewerGitGuardEnv({ PATH: process.env.PATH }, true); + expect(env.PATH).toContain('ejclaw-reviewer-git-'); + expect(env.EJCLAW_REAL_GIT).toBeTruthy(); + }); + + it('blocks mutating git subcommands even when git options come first', () => { + const env = buildReviewerGitGuardEnv({ PATH: process.env.PATH }, true); + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-reviewer-test-')); + + try { + execFileSync('git', ['-c', 'color.ui=false', 'commit', '-m', 'x'], { + cwd, + env, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }); + throw new Error('expected git wrapper to block commit'); + } catch (error) { + const stderr = + error instanceof Error && 'stderr' in error + ? String((error as Error & { stderr?: string | Buffer }).stderr ?? '') + : ''; + expect(stderr).toContain( + 'EJClaw reviewer runtime blocks mutating git subcommands: commit', + ); + } + }); +}); diff --git a/src/db.test.ts b/src/db.test.ts index 645a4d0..466fb07 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -6,6 +6,10 @@ import { _initTestDatabase, claimServiceHandoff, completeServiceHandoffAndAdvanceTargetCursor, + createPairedApproval, + createPairedArtifact, + createPairedExecution, + createPairedTask, createTask, createServiceHandoff, createProducedWorkItem, @@ -22,10 +26,17 @@ import { getRegisteredAgentTypesForJid, getMessagesSince, getNewMessages, + getPairedExecutionById, + getPairedProject, + getPairedTaskById, + getPairedWorkspace, getRouterStateForService, isPairedRoomJid, getSession, getTaskById, + listPairedApprovalsForTask, + listPairedArtifactsForTask, + listPairedWorkspacesForTask, markWorkItemDelivered, markWorkItemDeliveryRetry, setSession, @@ -33,6 +44,10 @@ import { setRouterStateForService, storeChatMetadata, storeMessage, + updatePairedExecution, + updatePairedTask, + upsertPairedProject, + upsertPairedWorkspace, updateTask, } from './db.js'; import { @@ -529,6 +544,156 @@ Check the run. }); }); +describe('paired task state', () => { + it('stores project, task, execution, workspace, approval, and artifact state', () => { + upsertPairedProject({ + chat_jid: 'dc:paired', + group_folder: 'paired-room', + canonical_work_dir: '/tmp/paired-room', + workspace_topology: 'shadow-snapshot', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + createPairedTask({ + id: 'paired-task-1', + chat_jid: 'dc:paired', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: 'wire up workspaces', + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + createPairedExecution({ + id: 'paired-exec-1', + task_id: 'paired-task-1', + service_id: 'codex-main', + role: 'owner', + workspace_id: null, + status: 'pending', + summary: null, + created_at: '2026-03-28T00:00:00.000Z', + started_at: null, + completed_at: null, + }); + + upsertPairedWorkspace({ + id: 'paired-task-1:owner', + task_id: 'paired-task-1', + role: 'owner', + workspace_dir: '/tmp/paired-room/owner', + snapshot_source_dir: null, + status: 'ready', + snapshot_refreshed_at: null, + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + const approvalId = createPairedApproval({ + task_id: 'paired-task-1', + service_id: 'codex-review', + role: 'reviewer', + status: 'pending', + note: 'needs snapshot', + created_at: '2026-03-28T00:00:01.000Z', + }); + + const artifactId = createPairedArtifact({ + task_id: 'paired-task-1', + execution_id: 'paired-exec-1', + service_id: 'codex-review', + artifact_type: 'report', + title: 'review notes', + content: 'looks fine', + file_path: null, + created_at: '2026-03-28T00:00:02.000Z', + }); + + expect(getPairedProject('dc:paired')?.canonical_work_dir).toBe( + '/tmp/paired-room', + ); + expect(getPairedTaskById('paired-task-1')?.status).toBe('draft'); + expect(getPairedExecutionById('paired-exec-1')?.status).toBe('pending'); + expect(getPairedWorkspace('paired-task-1', 'owner')?.workspace_dir).toBe( + '/tmp/paired-room/owner', + ); + expect(listPairedApprovalsForTask('paired-task-1')[0]?.id).toBe(approvalId); + expect(listPairedArtifactsForTask('paired-task-1')[0]?.id).toBe(artifactId); + }); + + it('updates task and execution state and keeps one workspace per role', () => { + createPairedTask({ + id: 'paired-task-2', + chat_jid: 'dc:paired', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: null, + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + createPairedExecution({ + id: 'paired-exec-2', + task_id: 'paired-task-2', + service_id: 'codex-review', + role: 'reviewer', + workspace_id: null, + status: 'pending', + summary: null, + created_at: '2026-03-28T00:00:00.000Z', + started_at: null, + completed_at: null, + }); + + updatePairedTask('paired-task-2', { + status: 'review_ready', + review_requested_at: '2026-03-28T00:10:00.000Z', + updated_at: '2026-03-28T00:10:00.000Z', + }); + updatePairedExecution('paired-exec-2', { + status: 'running', + started_at: '2026-03-28T00:11:00.000Z', + }); + + upsertPairedWorkspace({ + id: 'paired-task-2:reviewer', + task_id: 'paired-task-2', + role: 'reviewer', + workspace_dir: '/tmp/reviewer-v1', + snapshot_source_dir: '/tmp/owner', + status: 'ready', + snapshot_refreshed_at: '2026-03-28T00:10:00.000Z', + created_at: '2026-03-28T00:10:00.000Z', + updated_at: '2026-03-28T00:10:00.000Z', + }); + upsertPairedWorkspace({ + id: 'paired-task-2:reviewer', + task_id: 'paired-task-2', + role: 'reviewer', + workspace_dir: '/tmp/reviewer-v2', + snapshot_source_dir: '/tmp/owner', + status: 'ready', + snapshot_refreshed_at: '2026-03-28T00:12:00.000Z', + created_at: '2026-03-28T00:10:00.000Z', + updated_at: '2026-03-28T00:12:00.000Z', + }); + + expect(getPairedTaskById('paired-task-2')?.status).toBe('review_ready'); + expect(getPairedExecutionById('paired-exec-2')?.status).toBe('running'); + expect( + listPairedWorkspacesForTask('paired-task-2').map((workspace) => workspace.workspace_dir), + ).toEqual(['/tmp/reviewer-v2']); + }); +}); + // --- LIMIT behavior --- describe('message query LIMIT', () => { diff --git a/src/db.ts b/src/db.ts index 5ea0026..e88769a 100644 --- a/src/db.ts +++ b/src/db.ts @@ -25,6 +25,12 @@ import { getTaskRuntimeTaskId } from './task-watch-status.js'; import { NewMessage, AgentType, + PairedApproval, + PairedArtifact, + PairedExecution, + PairedProject, + PairedTask, + PairedWorkspace, RegisteredGroup, ScheduledTask, TaskRunLog, @@ -230,6 +236,100 @@ function createSchema(database: Database.Database): void { PRIMARY KEY (jid, agent_type), UNIQUE (folder, agent_type) ); + CREATE TABLE IF NOT EXISTS paired_projects ( + chat_jid TEXT PRIMARY KEY, + group_folder TEXT NOT NULL, + canonical_work_dir TEXT NOT NULL, + workspace_topology TEXT NOT NULL DEFAULT 'shadow-snapshot', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + CHECK (workspace_topology IN ('shadow-snapshot', 'reviewer-cow')) + ); + CREATE TABLE IF NOT EXISTS paired_tasks ( + id TEXT PRIMARY KEY, + chat_jid TEXT NOT NULL, + group_folder TEXT NOT NULL, + owner_service_id TEXT NOT NULL, + reviewer_service_id TEXT NOT NULL, + title TEXT, + source_ref TEXT, + review_requested_at TEXT, + status TEXT NOT NULL DEFAULT 'draft', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + CHECK ( + status IN ( + 'draft', + 'review_ready', + 'in_review', + 'changes_requested', + 'merge_ready', + 'merged', + 'discarded', + 'failed' + ) + ) + ); + CREATE INDEX IF NOT EXISTS idx_paired_tasks_chat_status + ON paired_tasks(chat_jid, status, updated_at); + CREATE TABLE IF NOT EXISTS paired_executions ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + service_id TEXT NOT NULL, + role TEXT NOT NULL, + workspace_id TEXT, + status TEXT NOT NULL DEFAULT 'pending', + summary TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + completed_at TEXT, + CHECK (role IN ('owner', 'reviewer')), + CHECK (status IN ('pending', 'running', 'succeeded', 'failed', 'cancelled')) + ); + CREATE INDEX IF NOT EXISTS idx_paired_executions_task + ON paired_executions(task_id, created_at); + CREATE TABLE IF NOT EXISTS paired_workspaces ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + role TEXT NOT NULL, + workspace_dir TEXT NOT NULL, + snapshot_source_dir TEXT, + status TEXT NOT NULL DEFAULT 'provisioning', + snapshot_refreshed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + CHECK (role IN ('owner', 'reviewer')), + CHECK (status IN ('provisioning', 'ready', 'stale', 'failed', 'archived')) + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_paired_workspaces_task_role + ON paired_workspaces(task_id, role); + CREATE TABLE IF NOT EXISTS paired_approvals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + service_id TEXT NOT NULL, + role TEXT NOT NULL, + status TEXT NOT NULL, + note TEXT, + created_at TEXT NOT NULL, + CHECK (role IN ('owner', 'reviewer')), + CHECK (status IN ('pending', 'approved', 'rejected')) + ); + CREATE INDEX IF NOT EXISTS idx_paired_approvals_task + ON paired_approvals(task_id, created_at); + CREATE TABLE IF NOT EXISTS paired_artifacts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + execution_id TEXT, + service_id TEXT NOT NULL, + artifact_type TEXT NOT NULL, + title TEXT, + content TEXT, + file_path TEXT, + created_at TEXT NOT NULL, + CHECK (artifact_type IN ('comment', 'report', 'patch')) + ); + CREATE INDEX IF NOT EXISTS idx_paired_artifacts_task + ON paired_artifacts(task_id, created_at); CREATE TABLE IF NOT EXISTS channel_owner ( chat_jid TEXT PRIMARY KEY, owner_service_id TEXT NOT NULL, @@ -1638,6 +1738,347 @@ export function isPairedRoomJid(jid: string): boolean { return types.includes('claude-code') && types.includes('codex'); } +// --- Paired task/project/workspace state --- + +export function upsertPairedProject(project: PairedProject): void { + db.prepare( + ` + INSERT INTO paired_projects ( + chat_jid, + group_folder, + canonical_work_dir, + workspace_topology, + created_at, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(chat_jid) DO UPDATE SET + group_folder = excluded.group_folder, + canonical_work_dir = excluded.canonical_work_dir, + workspace_topology = excluded.workspace_topology, + updated_at = excluded.updated_at + `, + ).run( + project.chat_jid, + project.group_folder, + project.canonical_work_dir, + project.workspace_topology, + project.created_at, + project.updated_at, + ); +} + +export function getPairedProject(chatJid: string): PairedProject | undefined { + return db + .prepare('SELECT * FROM paired_projects WHERE chat_jid = ?') + .get(chatJid) as PairedProject | undefined; +} + +export function createPairedTask(task: PairedTask): void { + db.prepare( + ` + INSERT INTO paired_tasks ( + id, + chat_jid, + group_folder, + owner_service_id, + reviewer_service_id, + title, + source_ref, + review_requested_at, + status, + created_at, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + ).run( + task.id, + task.chat_jid, + task.group_folder, + task.owner_service_id, + task.reviewer_service_id, + task.title, + task.source_ref, + task.review_requested_at, + task.status, + task.created_at, + task.updated_at, + ); +} + +export function getPairedTaskById(id: string): PairedTask | undefined { + return db + .prepare('SELECT * FROM paired_tasks WHERE id = ?') + .get(id) as PairedTask | undefined; +} + +export function getLatestOpenPairedTaskForChat( + chatJid: string, +): PairedTask | undefined { + return db + .prepare( + ` + SELECT * + FROM paired_tasks + WHERE chat_jid = ? + AND status NOT IN ('merged', 'discarded', 'failed') + ORDER BY updated_at DESC + LIMIT 1 + `, + ) + .get(chatJid) as PairedTask | undefined; +} + +export function updatePairedTask( + id: string, + updates: Partial< + Pick< + PairedTask, + 'title' | 'source_ref' | 'review_requested_at' | 'status' | 'updated_at' + > + >, +): void { + const fields: string[] = []; + const values: unknown[] = []; + + if (updates.title !== undefined) { + fields.push('title = ?'); + values.push(updates.title); + } + if (updates.source_ref !== undefined) { + fields.push('source_ref = ?'); + values.push(updates.source_ref); + } + if (updates.review_requested_at !== undefined) { + fields.push('review_requested_at = ?'); + values.push(updates.review_requested_at); + } + if (updates.status !== undefined) { + fields.push('status = ?'); + values.push(updates.status); + } + if (updates.updated_at !== undefined) { + fields.push('updated_at = ?'); + values.push(updates.updated_at); + } + + if (fields.length === 0) return; + + values.push(id); + db.prepare(`UPDATE paired_tasks SET ${fields.join(', ')} WHERE id = ?`).run( + ...values, + ); +} + +export function createPairedExecution(execution: PairedExecution): void { + db.prepare( + ` + INSERT INTO paired_executions ( + id, + task_id, + service_id, + role, + workspace_id, + status, + summary, + created_at, + started_at, + completed_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + ).run( + execution.id, + execution.task_id, + execution.service_id, + execution.role, + execution.workspace_id, + execution.status, + execution.summary, + execution.created_at, + execution.started_at, + execution.completed_at, + ); +} + +export function getPairedExecutionById( + id: string, +): PairedExecution | undefined { + return db + .prepare('SELECT * FROM paired_executions WHERE id = ?') + .get(id) as PairedExecution | undefined; +} + +export function updatePairedExecution( + id: string, + updates: Partial< + Pick< + PairedExecution, + 'workspace_id' | 'status' | 'summary' | 'started_at' | 'completed_at' + > + >, +): void { + const fields: string[] = []; + const values: unknown[] = []; + + if (updates.workspace_id !== undefined) { + fields.push('workspace_id = ?'); + values.push(updates.workspace_id); + } + if (updates.status !== undefined) { + fields.push('status = ?'); + values.push(updates.status); + } + if (updates.summary !== undefined) { + fields.push('summary = ?'); + values.push(updates.summary); + } + if (updates.started_at !== undefined) { + fields.push('started_at = ?'); + values.push(updates.started_at); + } + if (updates.completed_at !== undefined) { + fields.push('completed_at = ?'); + values.push(updates.completed_at); + } + + if (fields.length === 0) return; + + values.push(id); + db.prepare( + `UPDATE paired_executions SET ${fields.join(', ')} WHERE id = ?`, + ).run(...values); +} + +export function upsertPairedWorkspace(workspace: PairedWorkspace): void { + db.prepare( + ` + INSERT INTO paired_workspaces ( + id, + task_id, + role, + workspace_dir, + snapshot_source_dir, + status, + snapshot_refreshed_at, + created_at, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + workspace_dir = excluded.workspace_dir, + snapshot_source_dir = excluded.snapshot_source_dir, + status = excluded.status, + snapshot_refreshed_at = excluded.snapshot_refreshed_at, + updated_at = excluded.updated_at + `, + ).run( + workspace.id, + workspace.task_id, + workspace.role, + workspace.workspace_dir, + workspace.snapshot_source_dir, + workspace.status, + workspace.snapshot_refreshed_at, + workspace.created_at, + workspace.updated_at, + ); +} + +export function getPairedWorkspace( + taskId: string, + role: PairedWorkspace['role'], +): PairedWorkspace | undefined { + return db + .prepare('SELECT * FROM paired_workspaces WHERE task_id = ? AND role = ?') + .get(taskId, role) as PairedWorkspace | undefined; +} + +export function listPairedWorkspacesForTask(taskId: string): PairedWorkspace[] { + return db + .prepare( + 'SELECT * FROM paired_workspaces WHERE task_id = ? ORDER BY created_at', + ) + .all(taskId) as PairedWorkspace[]; +} + +export function createPairedApproval( + approval: Omit, +): number { + const result = db + .prepare( + ` + INSERT INTO paired_approvals ( + task_id, + service_id, + role, + status, + note, + created_at + ) + VALUES (?, ?, ?, ?, ?, ?) + `, + ) + .run( + approval.task_id, + approval.service_id, + approval.role, + approval.status, + approval.note, + approval.created_at, + ); + return Number(result.lastInsertRowid); +} + +export function listPairedApprovalsForTask(taskId: string): PairedApproval[] { + return db + .prepare( + 'SELECT * FROM paired_approvals WHERE task_id = ? ORDER BY created_at, id', + ) + .all(taskId) as PairedApproval[]; +} + +export function createPairedArtifact( + artifact: Omit, +): number { + const result = db + .prepare( + ` + INSERT INTO paired_artifacts ( + task_id, + execution_id, + service_id, + artifact_type, + title, + content, + file_path, + created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, + ) + .run( + artifact.task_id, + artifact.execution_id, + artifact.service_id, + artifact.artifact_type, + artifact.title, + artifact.content, + artifact.file_path, + artifact.created_at, + ); + return Number(result.lastInsertRowid); +} + +export function listPairedArtifactsForTask(taskId: string): PairedArtifact[] { + return db + .prepare( + 'SELECT * FROM paired_artifacts WHERE task_id = ? ORDER BY created_at, id', + ) + .all(taskId) as PairedArtifact[]; +} + /** * Get the most recent bot message (is_bot_message=1) in a chat, regardless of which bot sent it. * Used for duplicate detection in pair rooms. diff --git a/src/group-folder.ts b/src/group-folder.ts index 51fffcf..e9971f3 100644 --- a/src/group-folder.ts +++ b/src/group-folder.ts @@ -129,3 +129,32 @@ export function resolveServiceTaskSessionsPath( ensureWithinBase(sessionsBaseDir, sessionsPath); return sessionsPath; } + +export function resolvePairedTaskWorkspaceRoot( + folder: string, + taskId: string, +): string { + assertValidGroupFolder(folder); + assertValidRuntimeSegment(taskId, 'task ID'); + const workspacesBaseDir = path.resolve(DATA_DIR, 'workspaces'); + const workspacesPath = path.resolve( + workspacesBaseDir, + folder, + 'tasks', + taskId, + ); + ensureWithinBase(workspacesBaseDir, workspacesPath); + return workspacesPath; +} + +export function resolvePairedTaskWorkspacePath( + folder: string, + taskId: string, + role: 'owner' | 'reviewer', +): string { + assertValidRuntimeSegment(role, 'workspace role'); + const workspaceRoot = resolvePairedTaskWorkspaceRoot(folder, taskId); + const workspacePath = path.resolve(workspaceRoot, role); + ensureWithinBase(workspaceRoot, workspacePath); + return workspacePath; +} diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 54a6dc8..7970dfb 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -108,11 +108,18 @@ vi.mock('./memento-client.js', () => ({ buildRoomMemoryBriefing: vi.fn(), })); +vi.mock('./paired-execution-context.js', () => ({ + completePairedExecutionContext: vi.fn(), + markRoomReviewReady: vi.fn(), + preparePairedExecutionContext: vi.fn(() => undefined), +})); + import * as agentRunner from './agent-runner.js'; import * as codexTokenRotation from './codex-token-rotation.js'; import * as db from './db.js'; import { buildRoomMemoryBriefing } from './memento-client.js'; import { runAgentForGroup } from './message-agent-executor.js'; +import * as pairedExecutionContext from './paired-execution-context.js'; import * as sessionRecovery from './session-recovery.js'; import * as serviceRouting from './service-routing.js'; import * as tokenRotation from './token-rotation.js'; @@ -180,6 +187,7 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Function), undefined, + undefined, ); }); @@ -208,6 +216,7 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Function), undefined, + undefined, ); }); @@ -231,6 +240,7 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Function), undefined, + undefined, ); }); @@ -257,6 +267,7 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Function), undefined, + undefined, ); }); @@ -288,8 +299,179 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Function), undefined, + undefined, ); }); + + it('passes paired workspace env overrides into the runner when execution metadata exists', async () => { + const group = { + ...makeGroup(), + folder: 'test-group', + workDir: '/repo/canonical', + }; + + vi.mocked(pairedExecutionContext.preparePairedExecutionContext).mockReturnValue( + { + task: { + id: 'paired-task-1', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }, + execution: { + id: 'run-room-role:claude', + task_id: 'paired-task-1', + service_id: 'claude', + role: 'owner', + workspace_id: 'paired-task-1:owner', + status: 'running', + summary: null, + created_at: '2026-03-28T00:00:00.000Z', + started_at: '2026-03-28T00:00:00.000Z', + completed_at: null, + }, + workspace: { + id: 'paired-task-1:owner', + task_id: 'paired-task-1', + role: 'owner', + workspace_dir: '/tmp/paired/owner', + snapshot_source_dir: null, + status: 'ready', + snapshot_refreshed_at: null, + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }, + envOverrides: { + EJCLAW_WORK_DIR: '/tmp/paired/owner', + EJCLAW_PAIRED_TASK_ID: 'paired-task-1', + EJCLAW_PAIRED_EXECUTION_ID: 'run-room-role:claude', + EJCLAW_PAIRED_ROLE: 'owner', + }, + }, + ); + + await runAgentForGroup(makeDeps(), { + group, + prompt: 'hello', + chatJid: 'group@test', + runId: 'run-room-role', + }); + + expect(agentRunner.runAgentProcess).toHaveBeenCalledWith( + group, + expect.objectContaining({ + roomRoleContext: expect.any(Object), + }), + expect.any(Function), + undefined, + expect.objectContaining({ + EJCLAW_WORK_DIR: '/tmp/paired/owner', + EJCLAW_PAIRED_TASK_ID: 'paired-task-1', + EJCLAW_PAIRED_EXECUTION_ID: 'run-room-role:claude', + EJCLAW_PAIRED_ROLE: 'owner', + }), + ); + expect( + pairedExecutionContext.completePairedExecutionContext, + ).toHaveBeenCalledWith({ + executionId: 'run-room-role:claude', + status: 'succeeded', + summary: 'ok', + }); + }); + + it('blocks reviewer execution before review-ready and does not spawn the runner', async () => { + const group = { + ...makeGroup(), + folder: 'test-group', + workDir: '/repo/canonical', + }; + const outputs: Array<{ text?: string; result?: string | null }> = []; + + vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({ + chat_jid: 'group@test', + owner_service_id: 'codex-main', + reviewer_service_id: 'claude', + activated_at: null, + reason: null, + explicit: false, + }); + vi.mocked(pairedExecutionContext.preparePairedExecutionContext).mockReturnValue( + { + task: { + id: 'paired-task-1', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'codex-main', + reviewer_service_id: 'claude', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }, + execution: { + id: 'run-blocked-reviewer:claude', + task_id: 'paired-task-1', + service_id: 'claude', + role: 'reviewer', + workspace_id: null, + status: 'running', + summary: null, + created_at: '2026-03-28T00:00:00.000Z', + started_at: '2026-03-28T00:00:00.000Z', + completed_at: null, + }, + workspace: null, + envOverrides: { + EJCLAW_PAIRED_TASK_ID: 'paired-task-1', + EJCLAW_PAIRED_EXECUTION_ID: 'run-blocked-reviewer:claude', + EJCLAW_PAIRED_ROLE: 'reviewer', + EJCLAW_REVIEWER_RUNTIME: '1', + }, + blockMessage: + 'Review snapshot is not ready yet. Ask the owner to run /review-ready after preparing changes.', + }, + ); + + const result = await runAgentForGroup(makeDeps(), { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-blocked-reviewer', + onOutput: async (output) => { + outputs.push({ + text: output.output && 'text' in output.output ? output.output.text : undefined, + result: output.result, + }); + }, + }); + + expect(result).toBe('success'); + expect(agentRunner.runAgentProcess).not.toHaveBeenCalled(); + expect(outputs).toEqual([ + { + text: 'Review snapshot is not ready yet. Ask the owner to run /review-ready after preparing changes.', + result: null, + }, + ]); + expect( + pairedExecutionContext.completePairedExecutionContext, + ).toHaveBeenCalledWith({ + executionId: 'run-blocked-reviewer:claude', + status: 'failed', + summary: + 'Review snapshot is not ready yet. Ask the owner to run /review-ready after preparing changes.', + }); + }); }); describe('runAgentForGroup Claude rotation', () => { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index d68f020..6635deb 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -12,6 +12,10 @@ import { createServiceHandoff, getAllTasks } from './db.js'; import { GroupQueue } from './group-queue.js'; import { logger } from './logger.js'; import { buildRoomMemoryBriefing } from './memento-client.js'; +import { + completePairedExecutionContext, + preparePairedExecutionContext, +} from './paired-execution-context.js'; import { buildRoomRoleContext } from './room-role-context.js'; import { classifyRotationTrigger, @@ -126,9 +130,17 @@ export async function runAgentForGroup( currentLease, SERVICE_SESSION_SCOPE, ); + const pairedExecutionContext = preparePairedExecutionContext({ + group, + chatJid, + runId, + roomRoleContext, + }); const effectivePrompt = buildStructuredOutputPrompt(prompt, { reviewerMode, }); + let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; + let pairedExecutionSummary: string | null = null; const shouldHandoffToCodex = ( reason: AgentTriggerReason, @@ -188,6 +200,36 @@ export async function runAgentForGroup( roomRoleContext, }; + if (pairedExecutionContext?.blockMessage) { + pairedExecutionSummary = pairedExecutionContext.blockMessage.slice(0, 500); + logger.warn( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + serviceId: roomRoleContext?.serviceId, + role: roomRoleContext?.role, + }, + 'Blocked reviewer execution before review-ready snapshot was available', + ); + await onOutput?.({ + status: 'success', + result: null, + output: { + visibility: 'public', + text: pairedExecutionContext.blockMessage, + }, + phase: 'final', + }); + completePairedExecutionContext({ + executionId: pairedExecutionContext.execution.id, + status: pairedExecutionStatus, + summary: pairedExecutionSummary, + }); + return 'success'; + } + const runAttempt = async ( provider: string, ): Promise<{ @@ -237,6 +279,11 @@ export async function runAgentForGroup( streamedState = evaluation.state; const outputText = getAgentOutputText(output); + if (typeof outputText === 'string' && outputText.length > 0) { + pairedExecutionSummary = outputText.slice(0, 500); + } else if (typeof output.error === 'string' && output.error.length > 0) { + pairedExecutionSummary = output.error.slice(0, 500); + } if ( evaluation.newTrigger && typeof outputText === 'string' && @@ -345,6 +392,7 @@ export async function runAgentForGroup( (proc, processName, ipcDir) => deps.queue.registerProcess(chatJid, proc, processName, ipcDir), wrappedOnOutput, + pairedExecutionContext?.envOverrides, ); if (provider === 'claude' && output.newSessionId) { @@ -545,7 +593,8 @@ export async function runAgentForGroup( const provider = 'claude'; - let primaryAttempt = await runAttempt(provider); + try { + let primaryAttempt = await runAttempt(provider); const isRetryableClaudeSessionFailure = ( attempt: Awaited>, @@ -605,6 +654,9 @@ export async function runAgentForGroup( if (result === 'error') { return maybeHandoffAfterError(trigger.reason, primaryAttempt); } + if (result === 'success') { + pairedExecutionStatus = 'succeeded'; + } return result; } } @@ -613,7 +665,14 @@ export async function runAgentForGroup( const errMsg = getErrorMessage(primaryAttempt.error); const trigger = detectCodexRotationTrigger(errMsg); if (trigger.shouldRotate && getCodexAccountCount() > 1) { - return retryCodexWithRotation({ reason: trigger.reason }, errMsg); + const result = await retryCodexWithRotation( + { reason: trigger.reason }, + errMsg, + ); + if (result === 'success') { + pairedExecutionStatus = 'succeeded'; + } + return result; } } @@ -646,6 +705,17 @@ export async function runAgentForGroup( return 'error'; } + if (!pairedExecutionSummary) { + const finalOutputText = getAgentOutputText(output); + pairedExecutionSummary = + (typeof finalOutputText === 'string' && finalOutputText.length > 0 + ? finalOutputText.slice(0, 500) + : null) ?? + (typeof output.error === 'string' && output.error.length > 0 + ? output.error.slice(0, 500) + : null); + } + if ( canRotateToken && provider === 'claude' && @@ -697,6 +767,9 @@ export async function runAgentForGroup( if (result === 'error') { return maybeHandoffAfterError(trigger.reason, primaryAttempt); } + if (result === 'success') { + pairedExecutionStatus = 'succeeded'; + } return result; } } @@ -704,10 +777,14 @@ export async function runAgentForGroup( if (!isClaudeCodeAgent && getCodexAccountCount() > 1) { const trigger = detectCodexRotationTrigger(output.error); if (trigger.shouldRotate) { - return retryCodexWithRotation( + const result = await retryCodexWithRotation( { reason: trigger.reason }, output.error ?? undefined, ); + if (result === 'success') { + pairedExecutionStatus = 'succeeded'; + } + return result; } } @@ -729,13 +806,17 @@ export async function runAgentForGroup( primaryAttempt.streamedTriggerReason && getCodexAccountCount() > 1 ) { - return retryCodexWithRotation( + const result = await retryCodexWithRotation( { reason: primaryAttempt.streamedTriggerReason .reason as CodexRotationReason, }, output.error ?? output.result ?? undefined, ); + if (result === 'success') { + pairedExecutionStatus = 'succeeded'; + } + return result; } // Unresolved streamed trigger — rotation was unavailable or output was @@ -771,5 +852,15 @@ export async function runAgentForGroup( return 'error'; } - return 'success'; + pairedExecutionStatus = 'succeeded'; + return 'success'; + } finally { + if (pairedExecutionContext) { + completePairedExecutionContext({ + executionId: pairedExecutionContext.execution.id, + status: pairedExecutionStatus, + summary: pairedExecutionSummary, + }); + } + } } diff --git a/src/message-runtime.ts b/src/message-runtime.ts index eedfc8c..815201c 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -41,6 +41,8 @@ import { import { runAgentForGroup } from './message-agent-executor.js'; import { MessageTurnController } from './message-turn-controller.js'; import { createSuppressToken } from './output-suppression.js'; +import { markRoomReviewReady } from './paired-execution-context.js'; +import { buildRoomRoleContext } from './room-role-context.js'; import { extractSessionCommand, handleSessionCommand, @@ -50,7 +52,10 @@ import { import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; import { resolveGroupIpcPath } from './group-folder.js'; -import { shouldServiceProcessChat } from './service-routing.js'; +import { + getEffectiveChannelLease, + shouldServiceProcessChat, +} from './service-routing.js'; /** * Check if a message is a duplicate of the last bot final message in a paired room. @@ -557,6 +562,27 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { isTriggerAllowed(chatJid, msg.sender, loadSenderAllowlist()))) ); }, + markReviewReady: async () => { + const lease = getEffectiveChannelLease(chatJid); + const roomRoleContext = buildRoomRoleContext( + lease, + lease.owner_service_id, + ); + const result = markRoomReviewReady({ + group, + chatJid, + roomRoleContext, + }); + if (!result) { + return null; + } + return [ + 'Review snapshot updated.', + `- Task: ${result.task.id}`, + `- Owner workspace: ${result.ownerWorkspace.workspace_dir}`, + `- Reviewer snapshot: ${result.reviewerWorkspace.workspace_dir}`, + ].join('\n'); + }, }, }); if (cmdResult.handled) return cmdResult.success; diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts new file mode 100644 index 0000000..fbdaf6c --- /dev/null +++ b/src/paired-execution-context.test.ts @@ -0,0 +1,244 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./db.js', () => ({ + createPairedExecution: vi.fn(), + createPairedTask: vi.fn(), + getLatestOpenPairedTaskForChat: vi.fn(), + getPairedExecutionById: vi.fn(), + getPairedTaskById: vi.fn(), + getPairedWorkspace: vi.fn(), + updatePairedExecution: vi.fn(), + updatePairedTask: vi.fn(), + upsertPairedProject: vi.fn(), +})); + +vi.mock('./paired-workspace-manager.js', () => ({ + markPairedTaskReviewReady: vi.fn(), + provisionOwnerWorkspaceForPairedTask: vi.fn(), +})); + +vi.mock('./logger.js', () => ({ + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +import * as db from './db.js'; +import { + markRoomReviewReady, + preparePairedExecutionContext, +} from './paired-execution-context.js'; +import * as pairedWorkspaceManager from './paired-workspace-manager.js'; +import type { RegisteredGroup, RoomRoleContext } from './types.js'; + +const group: RegisteredGroup = { + name: 'Paired Room', + folder: 'paired-room', + trigger: '@codex', + added_at: '2026-03-28T00:00:00.000Z', + agentType: 'codex', + workDir: '/repo/canonical', +}; + +const ownerContext: RoomRoleContext = { + serviceId: 'codex-main', + role: 'owner', + ownerServiceId: 'codex-main', + reviewerServiceId: 'codex-review', + failoverOwner: false, +}; + +const reviewerContext: RoomRoleContext = { + serviceId: 'codex-review', + role: 'reviewer', + ownerServiceId: 'codex-main', + reviewerServiceId: 'codex-review', + failoverOwner: false, +}; + +describe('paired execution context', () => { + beforeEach(() => { + vi.resetAllMocks(); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue(undefined); + vi.mocked(db.getPairedExecutionById).mockReturnValue(undefined); + vi.mocked(db.getPairedTaskById).mockReturnValue(undefined); + vi.mocked(db.getPairedWorkspace).mockReturnValue(undefined); + vi.mocked( + pairedWorkspaceManager.provisionOwnerWorkspaceForPairedTask, + ).mockReturnValue({ + id: 'task-1:owner', + task_id: 'task-1', + role: 'owner', + workspace_dir: '/tmp/paired/task-1/owner', + snapshot_source_dir: null, + status: 'ready', + snapshot_refreshed_at: null, + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + }); + + it('creates an owner execution with a worktree override', () => { + const result = preparePairedExecutionContext({ + group, + chatJid: 'dc:test', + runId: 'run-1', + roomRoleContext: ownerContext, + }); + + expect(db.upsertPairedProject).toHaveBeenCalled(); + expect(db.createPairedTask).toHaveBeenCalledTimes(1); + expect(db.createPairedExecution).toHaveBeenCalledWith( + expect.objectContaining({ + id: 'run-1:codex-main', + role: 'owner', + workspace_id: 'task-1:owner', + }), + ); + expect(result?.envOverrides).toMatchObject({ + EJCLAW_WORK_DIR: '/tmp/paired/task-1/owner', + EJCLAW_PAIRED_ROLE: 'owner', + }); + }); + + it('uses the reviewer snapshot only after review_ready and marks the task in_review', () => { + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ + id: 'task-1', + chat_jid: 'dc:test', + group_folder: group.folder, + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: '2026-03-28T00:00:00.000Z', + status: 'review_ready', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + vi.mocked(db.getPairedWorkspace).mockReturnValue({ + id: 'task-1:reviewer', + task_id: 'task-1', + role: 'reviewer', + workspace_dir: '/tmp/paired/task-1/reviewer', + snapshot_source_dir: '/tmp/paired/task-1/owner', + status: 'ready', + snapshot_refreshed_at: '2026-03-28T00:00:00.000Z', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + const result = preparePairedExecutionContext({ + group, + chatJid: 'dc:test', + runId: 'run-2', + roomRoleContext: reviewerContext, + }); + + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ status: 'in_review' }), + ); + expect(result?.envOverrides).toMatchObject({ + EJCLAW_WORK_DIR: '/tmp/paired/task-1/reviewer', + EJCLAW_REVIEWER_RUNTIME: '1', + EJCLAW_PAIRED_ROLE: 'reviewer', + }); + }); + + it('blocks reviewer execution before review_ready instead of falling back to the canonical repo', () => { + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ + id: 'task-1', + chat_jid: 'dc:test', + group_folder: group.folder, + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + const result = preparePairedExecutionContext({ + group, + chatJid: 'dc:test', + runId: 'run-blocked-reviewer', + roomRoleContext: reviewerContext, + }); + + expect(result?.blockMessage).toBe( + 'Review snapshot is not ready yet. Ask the owner to run /review-ready after preparing changes.', + ); + expect(result?.envOverrides.EJCLAW_WORK_DIR).toBeUndefined(); + expect(db.getPairedWorkspace).not.toHaveBeenCalled(); + }); + + it('marks the active room task review-ready through the workspace manager', () => { + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ + id: 'task-1', + chat_jid: 'dc:test', + group_folder: group.folder, + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + vi.mocked(db.getPairedTaskById).mockReturnValue({ + id: 'task-1', + chat_jid: 'dc:test', + group_folder: group.folder, + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: '2026-03-28T00:00:00.000Z', + status: 'review_ready', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + vi.mocked(pairedWorkspaceManager.markPairedTaskReviewReady).mockReturnValue( + { + ownerWorkspace: { + id: 'task-1:owner', + task_id: 'task-1', + role: 'owner', + workspace_dir: '/tmp/paired/task-1/owner', + snapshot_source_dir: null, + status: 'ready', + snapshot_refreshed_at: null, + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }, + reviewerWorkspace: { + id: 'task-1:reviewer', + task_id: 'task-1', + role: 'reviewer', + workspace_dir: '/tmp/paired/task-1/reviewer', + snapshot_source_dir: '/tmp/paired/task-1/owner', + status: 'ready', + snapshot_refreshed_at: '2026-03-28T00:00:00.000Z', + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }, + }, + ); + + const result = markRoomReviewReady({ + group, + chatJid: 'dc:test', + roomRoleContext: ownerContext, + }); + + expect(pairedWorkspaceManager.markPairedTaskReviewReady).toHaveBeenCalledWith( + 'task-1', + ); + expect(result?.task.status).toBe('review_ready'); + }); +}); diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts new file mode 100644 index 0000000..a612944 --- /dev/null +++ b/src/paired-execution-context.ts @@ -0,0 +1,258 @@ +import { execFileSync } from 'child_process'; +import crypto from 'crypto'; + +import { + createPairedExecution, + createPairedTask, + getLatestOpenPairedTaskForChat, + getPairedExecutionById, + getPairedTaskById, + getPairedWorkspace, + updatePairedExecution, + updatePairedTask, + upsertPairedProject, +} from './db.js'; +import { logger } from './logger.js'; +import { + markPairedTaskReviewReady, + provisionOwnerWorkspaceForPairedTask, +} from './paired-workspace-manager.js'; +import type { + PairedExecution, + PairedTask, + PairedWorkspace, + RegisteredGroup, + RoomRoleContext, +} from './types.js'; + +function resolveCanonicalSourceRef(workDir: string): string { + try { + const head = execFileSync('git', ['rev-parse', 'HEAD'], { + cwd: workDir, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); + return head || 'HEAD'; + } catch { + return 'HEAD'; + } +} + +function ensurePairedProject( + group: RegisteredGroup, + chatJid: string, +): string | null { + if (!group.workDir) { + return null; + } + + const now = new Date().toISOString(); + upsertPairedProject({ + chat_jid: chatJid, + group_folder: group.folder, + canonical_work_dir: group.workDir, + workspace_topology: 'shadow-snapshot', + created_at: now, + updated_at: now, + }); + return group.workDir; +} + +function ensureActiveTask( + group: RegisteredGroup, + chatJid: string, + roomRoleContext: RoomRoleContext, +): PairedTask | null { + const canonicalWorkDir = ensurePairedProject(group, chatJid); + if (!canonicalWorkDir) { + return null; + } + + const existing = getLatestOpenPairedTaskForChat(chatJid); + if (existing) { + return existing; + } + + const now = new Date().toISOString(); + const task: PairedTask = { + id: crypto.randomUUID(), + chat_jid: chatJid, + group_folder: group.folder, + owner_service_id: roomRoleContext.ownerServiceId, + reviewer_service_id: roomRoleContext.reviewerServiceId, + title: null, + source_ref: resolveCanonicalSourceRef(canonicalWorkDir), + review_requested_at: null, + status: 'draft', + created_at: now, + updated_at: now, + }; + createPairedTask(task); + logger.info( + { + chatJid, + groupFolder: group.folder, + taskId: task.id, + sourceRef: task.source_ref, + }, + 'Created active paired task for room', + ); + return task; +} + +function ensureExecutionRecord(args: { + runId: string; + roomRoleContext: RoomRoleContext; + task: PairedTask; + workspace?: PairedWorkspace; +}): PairedExecution { + const executionId = `${args.runId}:${args.roomRoleContext.serviceId}`; + const existing = getPairedExecutionById(executionId); + const now = new Date().toISOString(); + + if (existing) { + updatePairedExecution(existing.id, { + workspace_id: args.workspace?.id ?? existing.workspace_id, + status: 'running', + started_at: existing.started_at ?? now, + }); + return { + ...existing, + workspace_id: args.workspace?.id ?? existing.workspace_id, + status: 'running', + started_at: existing.started_at ?? now, + }; + } + + const execution: PairedExecution = { + id: executionId, + task_id: args.task.id, + service_id: args.roomRoleContext.serviceId, + role: args.roomRoleContext.role, + workspace_id: args.workspace?.id ?? null, + status: 'running', + summary: null, + created_at: now, + started_at: now, + completed_at: null, + }; + createPairedExecution(execution); + return execution; +} + +export interface PreparedPairedExecutionContext { + task: PairedTask; + execution: PairedExecution; + workspace: PairedWorkspace | null; + envOverrides: Record; + blockMessage?: string; +} + +export function preparePairedExecutionContext(args: { + group: RegisteredGroup; + chatJid: string; + runId: string; + roomRoleContext?: RoomRoleContext; +}): PreparedPairedExecutionContext | undefined { + const { group, chatJid, runId, roomRoleContext } = args; + if (!roomRoleContext || !group.workDir) { + return undefined; + } + + const task = ensureActiveTask(group, chatJid, roomRoleContext); + if (!task) { + return undefined; + } + + let workspace: PairedWorkspace | null = null; + let blockMessage: string | undefined; + const now = new Date().toISOString(); + + if (roomRoleContext.role === 'owner') { + workspace = provisionOwnerWorkspaceForPairedTask(task.id); + } else if ( + task.status === 'review_ready' || + task.status === 'in_review' || + task.status === 'merge_ready' || + task.status === 'changes_requested' + ) { + workspace = getPairedWorkspace(task.id, 'reviewer') ?? null; + if (workspace && task.status === 'review_ready') { + updatePairedTask(task.id, { + status: 'in_review', + updated_at: now, + }); + } + } else { + blockMessage = + 'Review snapshot is not ready yet. Ask the owner to run /review-ready after preparing changes.'; + } + + const execution = ensureExecutionRecord({ + runId, + roomRoleContext, + task, + workspace: workspace ?? undefined, + }); + const envOverrides: Record = { + EJCLAW_PAIRED_TASK_ID: task.id, + EJCLAW_PAIRED_EXECUTION_ID: execution.id, + EJCLAW_PAIRED_ROLE: roomRoleContext.role, + }; + + if (workspace?.workspace_dir) { + envOverrides.EJCLAW_WORK_DIR = workspace.workspace_dir; + } + if (roomRoleContext.role === 'reviewer') { + envOverrides.EJCLAW_REVIEWER_RUNTIME = '1'; + } + + return { + task: getPairedTaskById(task.id) ?? task, + execution, + workspace, + envOverrides, + blockMessage, + }; +} + +export function completePairedExecutionContext(args: { + executionId: string; + status: 'succeeded' | 'failed'; + summary?: string | null; +}): void { + updatePairedExecution(args.executionId, { + status: args.status, + summary: args.summary ?? null, + completed_at: new Date().toISOString(), + }); +} + +export function markRoomReviewReady(args: { + group: RegisteredGroup; + chatJid: string; + roomRoleContext?: RoomRoleContext; +}): { + task: PairedTask; + ownerWorkspace: PairedWorkspace; + reviewerWorkspace: PairedWorkspace; +} | null { + const { group, chatJid, roomRoleContext } = args; + if (!roomRoleContext || !group.workDir) { + return null; + } + + const task = ensureActiveTask(group, chatJid, roomRoleContext); + if (!task) { + return null; + } + + const { ownerWorkspace, reviewerWorkspace } = markPairedTaskReviewReady( + task.id, + ); + return { + task: getPairedTaskById(task.id) ?? task, + ownerWorkspace, + reviewerWorkspace, + }; +} diff --git a/src/paired-workspace-manager.test.ts b/src/paired-workspace-manager.test.ts new file mode 100644 index 0000000..3fbb155 --- /dev/null +++ b/src/paired-workspace-manager.test.ts @@ -0,0 +1,274 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +function runGit(args: string[], cwd: string): string { + return execFileSync('git', args, { + cwd, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +async function loadModules() { + const db = await import('./db.js'); + const manager = await import('./paired-workspace-manager.js'); + return { db, manager }; +} + +describe('paired workspace manager', () => { + let tempRoot: string; + let previousDataDir: string | undefined; + let previousGroupsDir: string | undefined; + let previousStoreDir: string | undefined; + + beforeEach(() => { + tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-paired-workspace-')); + previousDataDir = process.env.EJCLAW_DATA_DIR; + previousGroupsDir = process.env.EJCLAW_GROUPS_DIR; + previousStoreDir = process.env.EJCLAW_STORE_DIR; + process.env.EJCLAW_DATA_DIR = path.join(tempRoot, 'data'); + process.env.EJCLAW_GROUPS_DIR = path.join(tempRoot, 'groups'); + process.env.EJCLAW_STORE_DIR = path.join(tempRoot, 'store'); + vi.resetModules(); + }); + + afterEach(() => { + if (previousDataDir === undefined) delete process.env.EJCLAW_DATA_DIR; + else process.env.EJCLAW_DATA_DIR = previousDataDir; + if (previousGroupsDir === undefined) delete process.env.EJCLAW_GROUPS_DIR; + else process.env.EJCLAW_GROUPS_DIR = previousGroupsDir; + if (previousStoreDir === undefined) delete process.env.EJCLAW_STORE_DIR; + else process.env.EJCLAW_STORE_DIR = previousStoreDir; + fs.rmSync(tempRoot, { recursive: true, force: true }); + }); + + it('provisions an owner worktree and refreshes a reviewer shadow snapshot', async () => { + const { db, manager } = await loadModules(); + db._initTestDatabase(); + + const canonicalDir = path.join(tempRoot, 'canonical'); + fs.mkdirSync(canonicalDir, { recursive: true }); + runGit(['init'], canonicalDir); + runGit(['config', 'user.email', 'test@example.com'], canonicalDir); + runGit(['config', 'user.name', 'EJClaw Test'], canonicalDir); + fs.writeFileSync(path.join(canonicalDir, 'README.md'), 'original\n'); + runGit(['add', 'README.md'], canonicalDir); + runGit(['commit', '-m', 'initial'], canonicalDir); + + const now = '2026-03-28T00:00:00.000Z'; + db.upsertPairedProject({ + chat_jid: 'dc:test', + group_folder: 'paired-room', + canonical_work_dir: canonicalDir, + workspace_topology: 'shadow-snapshot', + created_at: now, + updated_at: now, + }); + db.createPairedTask({ + id: 'paired-task-1', + chat_jid: 'dc:test', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: 'review the owner changes', + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: now, + updated_at: now, + }); + + const ownerWorkspace = + manager.provisionOwnerWorkspaceForPairedTask('paired-task-1'); + expect(fs.existsSync(path.join(ownerWorkspace.workspace_dir, '.git'))).toBe( + true, + ); + + fs.writeFileSync( + path.join(ownerWorkspace.workspace_dir, 'README.md'), + 'owner modified\n', + ); + fs.writeFileSync( + path.join(ownerWorkspace.workspace_dir, 'NEW_FILE.txt'), + 'review me\n', + ); + + const { reviewerWorkspace } = + manager.markPairedTaskReviewReady('paired-task-1'); + + expect( + fs.readFileSync(path.join(reviewerWorkspace.workspace_dir, 'README.md'), 'utf-8'), + ).toBe('owner modified\n'); + expect( + fs.readFileSync( + path.join(reviewerWorkspace.workspace_dir, 'NEW_FILE.txt'), + 'utf-8', + ), + ).toBe('review me\n'); + expect( + runGit(['config', '--local', '--get', 'remote.origin.pushurl'], reviewerWorkspace.workspace_dir), + ).toBe('DISABLED_BY_EJCLAW'); + expect( + runGit(['status', '--short'], reviewerWorkspace.workspace_dir), + ).toContain('M README.md'); + expect( + db.getPairedTaskById('paired-task-1')?.status, + ).toBe('review_ready'); + expect( + db.getPairedWorkspace('paired-task-1', 'reviewer')?.snapshot_refreshed_at, + ).toBeTruthy(); + }); + + it('replaces stale reviewer files on snapshot refresh', async () => { + const { db, manager } = await loadModules(); + db._initTestDatabase(); + + const canonicalDir = path.join(tempRoot, 'canonical'); + fs.mkdirSync(canonicalDir, { recursive: true }); + runGit(['init'], canonicalDir); + runGit(['config', 'user.email', 'test@example.com'], canonicalDir); + runGit(['config', 'user.name', 'EJClaw Test'], canonicalDir); + fs.writeFileSync(path.join(canonicalDir, 'keep.txt'), 'keep\n'); + fs.writeFileSync(path.join(canonicalDir, 'remove.txt'), 'remove\n'); + runGit(['add', 'keep.txt', 'remove.txt'], canonicalDir); + runGit(['commit', '-m', 'initial'], canonicalDir); + + const now = '2026-03-28T00:00:00.000Z'; + db.upsertPairedProject({ + chat_jid: 'dc:test', + group_folder: 'paired-room', + canonical_work_dir: canonicalDir, + workspace_topology: 'shadow-snapshot', + created_at: now, + updated_at: now, + }); + db.createPairedTask({ + id: 'paired-task-2', + chat_jid: 'dc:test', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: now, + updated_at: now, + }); + + const ownerWorkspace = + manager.provisionOwnerWorkspaceForPairedTask('paired-task-2'); + manager.refreshReviewerSnapshotForPairedTask('paired-task-2'); + + fs.rmSync(path.join(ownerWorkspace.workspace_dir, 'remove.txt')); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, 'keep.txt'), 'updated\n'); + + const reviewerWorkspace = + manager.refreshReviewerSnapshotForPairedTask('paired-task-2'); + + expect( + fs.existsSync(path.join(reviewerWorkspace.workspace_dir, 'remove.txt')), + ).toBe(false); + expect( + fs.readFileSync(path.join(reviewerWorkspace.workspace_dir, 'keep.txt'), 'utf-8'), + ).toBe('updated\n'); + expect( + runGit(['status', '--short'], reviewerWorkspace.workspace_dir), + ).toContain('D remove.txt'); + }); + + it('filters secrets, caches, and build outputs out of reviewer snapshots', async () => { + const { db, manager } = await loadModules(); + db._initTestDatabase(); + + const canonicalDir = path.join(tempRoot, 'canonical'); + fs.mkdirSync(canonicalDir, { recursive: true }); + runGit(['init'], canonicalDir); + runGit(['config', 'user.email', 'test@example.com'], canonicalDir); + runGit(['config', 'user.name', 'EJClaw Test'], canonicalDir); + fs.writeFileSync(path.join(canonicalDir, 'tracked.ts'), 'export const ok = 1;\n'); + fs.writeFileSync(path.join(canonicalDir, '.env.production'), 'TRACKED_SECRET=1\n'); + fs.writeFileSync(path.join(canonicalDir, '.env.example'), 'EXAMPLE=1\n'); + runGit(['add', 'tracked.ts', '.env.production', '.env.example'], canonicalDir); + runGit(['commit', '-m', 'initial'], canonicalDir); + + const now = '2026-03-28T00:00:00.000Z'; + db.upsertPairedProject({ + chat_jid: 'dc:test', + group_folder: 'paired-room', + canonical_work_dir: canonicalDir, + workspace_topology: 'shadow-snapshot', + created_at: now, + updated_at: now, + }); + db.createPairedTask({ + id: 'paired-task-3', + chat_jid: 'dc:test', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: null, + status: 'draft', + created_at: now, + updated_at: now, + }); + + const ownerWorkspace = + manager.provisionOwnerWorkspaceForPairedTask('paired-task-3'); + fs.mkdirSync(path.join(ownerWorkspace.workspace_dir, 'src'), { + recursive: true, + }); + fs.mkdirSync(path.join(ownerWorkspace.workspace_dir, 'node_modules', '.cache'), { + recursive: true, + }); + fs.mkdirSync(path.join(ownerWorkspace.workspace_dir, 'dist'), { + recursive: true, + }); + fs.mkdirSync(path.join(ownerWorkspace.workspace_dir, 'logs'), { + recursive: true, + }); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, 'src', 'draft.ts'), 'export const draft = true;\n'); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, '.env.local'), 'SECRET=1\n'); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, 'node_modules', '.cache', 'x'), 'cache\n'); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, 'dist', 'bundle.js'), 'dist\n'); + fs.writeFileSync(path.join(ownerWorkspace.workspace_dir, 'logs', 'debug.log'), 'log\n'); + + const reviewerWorkspace = + manager.refreshReviewerSnapshotForPairedTask('paired-task-3'); + + expect( + fs.readFileSync( + path.join(reviewerWorkspace.workspace_dir, 'src', 'draft.ts'), + 'utf-8', + ), + ).toBe('export const draft = true;\n'); + expect( + fs.existsSync(path.join(reviewerWorkspace.workspace_dir, '.env.production')), + ).toBe(false); + expect( + fs.readFileSync( + path.join(reviewerWorkspace.workspace_dir, '.env.example'), + 'utf-8', + ), + ).toBe('EXAMPLE=1\n'); + expect( + fs.existsSync(path.join(reviewerWorkspace.workspace_dir, '.env.local')), + ).toBe(false); + expect( + fs.existsSync( + path.join(reviewerWorkspace.workspace_dir, 'node_modules', '.cache', 'x'), + ), + ).toBe(false); + expect( + fs.existsSync(path.join(reviewerWorkspace.workspace_dir, 'dist', 'bundle.js')), + ).toBe(false); + expect( + fs.existsSync(path.join(reviewerWorkspace.workspace_dir, 'logs', 'debug.log')), + ).toBe(false); + }); +}); diff --git a/src/paired-workspace-manager.ts b/src/paired-workspace-manager.ts new file mode 100644 index 0000000..fd34287 --- /dev/null +++ b/src/paired-workspace-manager.ts @@ -0,0 +1,337 @@ +import { execFileSync } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +import { + getPairedProject, + getPairedTaskById, + getPairedWorkspace, + updatePairedTask, + upsertPairedWorkspace, +} from './db.js'; +import { resolvePairedTaskWorkspacePath } from './group-folder.js'; +import { logger } from './logger.js'; +import type { PairedTask, PairedWorkspace } from './types.js'; + +const REVIEWER_SNAPSHOT_DENY_SEGMENTS = new Set([ + '.git', + '.claude', + '.codex', + '.next', + '.turbo', + '.cache', + 'node_modules', + 'dist', + 'build', + 'coverage', + 'logs', +]); + +const REVIEWER_SNAPSHOT_ALLOWED_UNTRACKED_EXTENSIONS = new Set([ + '.c', + '.cc', + '.cpp', + '.css', + '.csv', + '.go', + '.graphql', + '.h', + '.hpp', + '.html', + '.java', + '.js', + '.json', + '.jsx', + '.kt', + '.md', + '.mjs', + '.prisma', + '.proto', + '.py', + '.rb', + '.rs', + '.scss', + '.sh', + '.sql', + '.svg', + '.swift', + '.toml', + '.ts', + '.tsx', + '.txt', + '.xml', + '.yaml', + '.yml', +]); + +const REVIEWER_SNAPSHOT_ALLOWED_UNTRACKED_BASENAMES = new Set([ + '.editorconfig', + '.eslintignore', + '.eslintrc', + '.gitattributes', + '.gitignore', + '.npmrc', + '.nvmrc', + '.prettierignore', + '.prettierrc', + 'Dockerfile', + 'Makefile', + 'README', + 'README.md', + 'package-lock.json', + 'pnpm-lock.yaml', + 'tsconfig.json', + 'yarn.lock', +]); + +function runGit(args: string[], cwd?: string): string { + return execFileSync('git', args, { + cwd, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +function ensureGitRepository(repoDir: string): void { + const insideWorkTree = runGit(['rev-parse', '--is-inside-work-tree'], repoDir); + if (insideWorkTree !== 'true') { + throw new Error(`Not a git repository: ${repoDir}`); + } +} + +function ensureCleanDirectory(targetDir: string): void { + fs.mkdirSync(targetDir, { recursive: true }); +} + +function resetDirectoryExceptGit(targetDir: string): void { + if (!fs.existsSync(targetDir)) return; + for (const entry of fs.readdirSync(targetDir)) { + if (entry === '.git') continue; + fs.rmSync(path.join(targetDir, entry), { recursive: true, force: true }); + } +} + +function listGitPaths(repoDir: string, args: string[]): string[] { + const output = execFileSync('git', args, { + cwd: repoDir, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }); + return output + .split('\0') + .map((value) => value.trim()) + .filter(Boolean); +} + +function isReviewerSnapshotDeniedPath(relativePath: string): boolean { + const segments = relativePath.split(/[\\/]+/).filter(Boolean); + if (segments.some((segment) => REVIEWER_SNAPSHOT_DENY_SEGMENTS.has(segment))) { + return true; + } + + const basename = path.basename(relativePath); + if (basename === '.env') { + return true; + } + if ( + basename.startsWith('.env.') && + basename !== '.env.example' && + basename !== '.env.sample' + ) { + return true; + } + if (basename.endsWith('.log')) { + return true; + } + + return false; +} + +function shouldIncludeUntrackedReviewerPath(relativePath: string): boolean { + if (isReviewerSnapshotDeniedPath(relativePath)) { + return false; + } + + const basename = path.basename(relativePath); + if (REVIEWER_SNAPSHOT_ALLOWED_UNTRACKED_BASENAMES.has(basename)) { + return true; + } + + return REVIEWER_SNAPSHOT_ALLOWED_UNTRACKED_EXTENSIONS.has( + path.extname(basename).toLowerCase(), + ); +} + +function copySelectedSnapshotTree(sourceDir: string, targetDir: string): void { + resetDirectoryExceptGit(targetDir); + + const trackedFiles = listGitPaths(sourceDir, [ + 'ls-files', + '--cached', + '-z', + ]).filter((relativePath) => !isReviewerSnapshotDeniedPath(relativePath)); + const untrackedFiles = listGitPaths(sourceDir, [ + 'ls-files', + '--others', + '--exclude-standard', + '-z', + ]).filter(shouldIncludeUntrackedReviewerPath); + + const filesToCopy = [...new Set([...trackedFiles, ...untrackedFiles])].sort(); + + for (const relativePath of filesToCopy) { + const sourcePath = path.join(sourceDir, relativePath); + if (!fs.existsSync(sourcePath)) continue; + + const targetPath = path.join(targetDir, relativePath); + fs.mkdirSync(path.dirname(targetPath), { recursive: true }); + fs.cpSync(sourcePath, targetPath, { force: true, recursive: true }); + } +} + +function configureReviewerGitIsolation(workspaceDir: string): void { + try { + runGit(['config', '--local', 'push.default', 'nothing'], workspaceDir); + runGit(['config', '--local', 'credential.helper', ''], workspaceDir); + runGit( + ['config', '--local', 'remote.origin.pushurl', 'DISABLED_BY_EJCLAW'], + workspaceDir, + ); + } catch (error) { + logger.warn( + { workspaceDir, error }, + 'Failed to apply reviewer git isolation settings', + ); + } +} + +function getTaskAndProject(taskId: string): { + task: PairedTask; + canonicalWorkDir: string; +} { + const task = getPairedTaskById(taskId); + if (!task) { + throw new Error(`Paired task not found: ${taskId}`); + } + + const project = getPairedProject(task.chat_jid); + if (!project) { + throw new Error(`Paired project not found for chat: ${task.chat_jid}`); + } + + return { task, canonicalWorkDir: project.canonical_work_dir }; +} + +function makeWorkspaceRecord(args: { + taskId: string; + role: PairedWorkspace['role']; + workspaceDir: string; + snapshotSourceDir?: string | null; + snapshotRefreshedAt?: string | null; + createdAt?: string; +}): PairedWorkspace { + const existing = getPairedWorkspace(args.taskId, args.role); + const now = new Date().toISOString(); + return { + id: existing?.id || `${args.taskId}:${args.role}`, + task_id: args.taskId, + role: args.role, + workspace_dir: args.workspaceDir, + snapshot_source_dir: args.snapshotSourceDir ?? existing?.snapshot_source_dir ?? null, + status: 'ready', + snapshot_refreshed_at: + args.snapshotRefreshedAt ?? existing?.snapshot_refreshed_at ?? null, + created_at: existing?.created_at || args.createdAt || now, + updated_at: now, + }; +} + +export function provisionOwnerWorkspaceForPairedTask( + taskId: string, +): PairedWorkspace { + const { task, canonicalWorkDir } = getTaskAndProject(taskId); + ensureGitRepository(canonicalWorkDir); + + const sourceRef = task.source_ref || 'HEAD'; + const workspaceDir = resolvePairedTaskWorkspacePath( + task.group_folder, + task.id, + 'owner', + ); + const parentDir = path.dirname(workspaceDir); + fs.mkdirSync(parentDir, { recursive: true }); + + if (!fs.existsSync(path.join(workspaceDir, '.git'))) { + ensureCleanDirectory(parentDir); + runGit(['worktree', 'add', '--detach', workspaceDir, sourceRef], canonicalWorkDir); + logger.info( + { taskId, workspaceDir, sourceRef }, + 'Provisioned owner workspace for paired task', + ); + } + + const workspace = makeWorkspaceRecord({ + taskId, + role: 'owner', + workspaceDir, + }); + upsertPairedWorkspace(workspace); + return workspace; +} + +export function refreshReviewerSnapshotForPairedTask( + taskId: string, +): PairedWorkspace { + const ownerWorkspace = provisionOwnerWorkspaceForPairedTask(taskId); + ensureGitRepository(ownerWorkspace.workspace_dir); + + const { task } = getTaskAndProject(taskId); + const reviewerDir = resolvePairedTaskWorkspacePath( + task.group_folder, + task.id, + 'reviewer', + ); + fs.mkdirSync(path.dirname(reviewerDir), { recursive: true }); + + if (!fs.existsSync(path.join(reviewerDir, '.git'))) { + fs.rmSync(reviewerDir, { recursive: true, force: true }); + runGit(['clone', '--shared', ownerWorkspace.workspace_dir, reviewerDir]); + } + + runGit(['reset', '--hard', 'HEAD'], reviewerDir); + runGit(['clean', '-fdx'], reviewerDir); + copySelectedSnapshotTree(ownerWorkspace.workspace_dir, reviewerDir); + configureReviewerGitIsolation(reviewerDir); + + const refreshedAt = new Date().toISOString(); + const workspace = makeWorkspaceRecord({ + taskId, + role: 'reviewer', + workspaceDir: reviewerDir, + snapshotSourceDir: ownerWorkspace.workspace_dir, + snapshotRefreshedAt: refreshedAt, + }); + upsertPairedWorkspace(workspace); + logger.info( + { taskId, reviewerDir, snapshotSourceDir: ownerWorkspace.workspace_dir }, + 'Refreshed reviewer snapshot for paired task', + ); + return workspace; +} + +export function markPairedTaskReviewReady(taskId: string): { + ownerWorkspace: PairedWorkspace; + reviewerWorkspace: PairedWorkspace; +} { + const ownerWorkspace = provisionOwnerWorkspaceForPairedTask(taskId); + const reviewerWorkspace = refreshReviewerSnapshotForPairedTask(taskId); + const now = new Date().toISOString(); + + updatePairedTask(taskId, { + status: 'review_ready', + review_requested_at: now, + updated_at: now, + }); + + return { ownerWorkspace, reviewerWorkspace }; +} diff --git a/src/session-commands.test.ts b/src/session-commands.test.ts index 436fa8a..046de1c 100644 --- a/src/session-commands.test.ts +++ b/src/session-commands.test.ts @@ -23,6 +23,12 @@ describe('extractSessionCommand', () => { expect(extractSessionCommand('/clear', trigger)).toBe('/clear'); }); + it('detects bare /review-ready', () => { + expect(extractSessionCommand('/review-ready', trigger)).toBe( + '/review-ready', + ); + }); + it('rejects /compact with extra text', () => { expect(extractSessionCommand('/compact now please', trigger)).toBeNull(); }); @@ -120,6 +126,7 @@ function makeDeps( formatMessages: vi.fn().mockReturnValue(''), isAdminSender: vi.fn().mockReturnValue(false), canSenderInteract: vi.fn().mockReturnValue(true), + markReviewReady: vi.fn().mockResolvedValue('Review snapshot updated.'), ...overrides, }; } @@ -178,6 +185,31 @@ describe('handleSessionCommand', () => { ); }); + it('handles authorized /review-ready without invoking the agent', async () => { + const deps = makeDeps({ + markReviewReady: vi.fn().mockResolvedValue( + 'Review snapshot updated.\n- Task: paired-task-1', + ), + }); + + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/review-ready')], + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + + expect(result).toEqual({ handled: true, success: true }); + expect(deps.markReviewReady).toHaveBeenCalledTimes(1); + expect(deps.runAgent).not.toHaveBeenCalled(); + expect(deps.advanceCursor).toHaveBeenCalledWith('100'); + expect(deps.sendMessage).toHaveBeenCalledWith( + 'Review snapshot updated.\n- Task: paired-task-1', + ); + }); + it('sends denial to interactable sender in non-main group', async () => { const deps = makeDeps(); const result = await handleSessionCommand({ diff --git a/src/session-commands.ts b/src/session-commands.ts index 50c38df..1ced58e 100644 --- a/src/session-commands.ts +++ b/src/session-commands.ts @@ -10,6 +10,8 @@ const SESSION_COMMAND_CONTROL_PATTERNS = [ /^Failed to process messages before \/compact\. Try again\.$/, /^\/compact failed\. The session is unchanged\.$/, /^Conversation compacted\.$/, + /^Review snapshot updated\.$/, + /^Review-ready is unavailable for this room\./, ]; /** @@ -24,6 +26,7 @@ export function extractSessionCommand( text = text.replace(triggerPattern, '').trim(); if (text === '/compact') return '/compact'; if (text === '/clear') return '/clear'; + if (text === '/review-ready') return '/review-ready'; return null; } @@ -68,6 +71,7 @@ export interface SessionCommandDeps { isAdminSender: (msg: NewMessage) => boolean; /** Whether the denied sender would normally be allowed to interact (for denial messages). */ canSenderInteract: (msg: NewMessage) => boolean; + markReviewReady: () => Promise; } function resultToText(result: string | object | null | undefined): string { @@ -149,6 +153,16 @@ export async function handleSessionCommand(opts: { return { handled: true, success: true }; } + if (command === '/review-ready') { + const result = await deps.markReviewReady(); + deps.advanceCursor(cmdMsg.timestamp); + await deps.sendMessage( + result ?? + 'Review-ready is unavailable for this room. Paired workspaces require a configured project workDir.', + ); + return { handled: true, success: true }; + } + const cmdIndex = missedMessages.indexOf(cmdMsg); const preCompactMsgs = missedMessages.slice(0, cmdIndex); diff --git a/src/types.ts b/src/types.ts index f5c27a3..549e507 100644 --- a/src/types.ts +++ b/src/types.ts @@ -25,6 +25,38 @@ export type AgentVisibility = 'public' | 'silent'; export type PairedRoomRole = 'owner' | 'reviewer'; +export type PairedWorkspaceTopology = 'shadow-snapshot' | 'reviewer-cow'; + +export type PairedTaskStatus = + | 'draft' + | 'review_ready' + | 'in_review' + | 'changes_requested' + | 'merge_ready' + | 'merged' + | 'discarded' + | 'failed'; + +export type PairedExecutionStatus = + | 'pending' + | 'running' + | 'succeeded' + | 'failed' + | 'cancelled'; + +export type PairedWorkspaceRole = 'owner' | 'reviewer'; + +export type PairedWorkspaceStatus = + | 'provisioning' + | 'ready' + | 'stale' + | 'failed' + | 'archived'; + +export type PairedApprovalStatus = 'pending' | 'approved' | 'rejected'; + +export type PairedArtifactType = 'comment' | 'report' | 'patch'; + export interface RoomRoleContext { serviceId: string; role: PairedRoomRole; @@ -33,6 +65,76 @@ export interface RoomRoleContext { failoverOwner: boolean; } +export interface PairedProject { + chat_jid: string; + group_folder: string; + canonical_work_dir: string; + workspace_topology: PairedWorkspaceTopology; + created_at: string; + updated_at: string; +} + +export interface PairedTask { + id: string; + chat_jid: string; + group_folder: string; + owner_service_id: string; + reviewer_service_id: string; + title: string | null; + source_ref: string | null; + review_requested_at: string | null; + status: PairedTaskStatus; + created_at: string; + updated_at: string; +} + +export interface PairedExecution { + id: string; + task_id: string; + service_id: string; + role: PairedWorkspaceRole; + workspace_id: string | null; + status: PairedExecutionStatus; + summary: string | null; + created_at: string; + started_at: string | null; + completed_at: string | null; +} + +export interface PairedWorkspace { + id: string; + task_id: string; + role: PairedWorkspaceRole; + workspace_dir: string; + snapshot_source_dir: string | null; + status: PairedWorkspaceStatus; + snapshot_refreshed_at: string | null; + created_at: string; + updated_at: string; +} + +export interface PairedApproval { + id: number; + task_id: string; + service_id: string; + role: PairedWorkspaceRole; + status: PairedApprovalStatus; + note: string | null; + created_at: string; +} + +export interface PairedArtifact { + id: number; + task_id: string; + execution_id: string | null; + service_id: string; + artifact_type: PairedArtifactType; + title: string | null; + content: string | null; + file_path: string | null; + created_at: string; +} + export type StructuredAgentOutput = | { visibility: 'public';