From 497e6dec749b049b10b2d2bac5769b9252b8e194 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Fri, 10 Apr 2026 22:48:49 +0900 Subject: [PATCH] refactor: decompose runtime coordination and runner internals (PR6) --- runners/agent-runner/bun.lock | 3 + runners/agent-runner/package.json | 1 + runners/agent-runner/src/reviewer-runtime.ts | 276 +--------- runners/agent-runner/src/room-role-context.ts | 27 +- runners/codex-runner/bun.lock | 3 + runners/codex-runner/package.json | 1 + runners/codex-runner/src/reviewer-runtime.ts | 260 +-------- runners/codex-runner/src/room-role-context.ts | 27 +- runners/shared/bun.lock | 20 + runners/shared/package.json | 15 + runners/shared/src/index.ts | 9 + runners/shared/src/reviewer-git-guard.ts | 254 +++++++++ runners/shared/src/room-role-context.ts | 23 + runners/shared/tsconfig.json | 15 + src/agent-runner-process.ts | 428 +++++++++++++++ src/agent-runner.ts | 425 +-------------- src/db.ts | 469 ++-------------- src/db/database-lifecycle.ts | 83 +++ src/db/rooms.ts | 499 ++++++++++++++++++ src/index.ts | 142 +---- src/message-runtime-turns.ts | 267 ++++++++++ src/message-runtime.ts | 411 +++++---------- src/runtime-state.ts | 133 +++++ 23 files changed, 1987 insertions(+), 1804 deletions(-) create mode 100644 runners/shared/bun.lock create mode 100644 runners/shared/package.json create mode 100644 runners/shared/src/index.ts create mode 100644 runners/shared/src/reviewer-git-guard.ts create mode 100644 runners/shared/src/room-role-context.ts create mode 100644 runners/shared/tsconfig.json create mode 100644 src/agent-runner-process.ts create mode 100644 src/db/database-lifecycle.ts create mode 100644 src/db/rooms.ts create mode 100644 src/message-runtime-turns.ts create mode 100644 src/runtime-state.ts diff --git a/runners/agent-runner/bun.lock b/runners/agent-runner/bun.lock index 1f55d2c..1621000 100644 --- a/runners/agent-runner/bun.lock +++ b/runners/agent-runner/bun.lock @@ -8,6 +8,7 @@ "@anthropic-ai/claude-agent-sdk": "^0.2.87", "@modelcontextprotocol/sdk": "^1.12.1", "cron-parser": "^5.0.0", + "ejclaw-runners-shared": "file:../shared", "zod": "^4.0.0", }, "devDependencies": { @@ -97,6 +98,8 @@ "ee-first": ["ee-first@1.1.1", "", {}, "sha512-WMwm9LhRUo+WUaRN+vRuETqG89IgZphVSNkdFgeb6sS/E4OrDIN7t48CAewSHXc6C8lefD8KKfr5vY61brQlow=="], + "ejclaw-runners-shared": ["ejclaw-runners-shared@file:../shared", { "devDependencies": { "@types/node": "^22.10.7", "typescript": "^5.7.3" } }], + "encodeurl": ["encodeurl@2.0.0", "", {}, "sha512-Q0n9HRi4m6JuGIV1eFlmvJB7ZEVxu93IrMyiMsGC0lrMJMWzRgx6WGquyfQgZVb31vhGgXnfmPNNXmxnOkRBrg=="], "es-define-property": ["es-define-property@1.0.1", "", {}, "sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g=="], diff --git a/runners/agent-runner/package.json b/runners/agent-runner/package.json index cd5a17b..6d6c718 100644 --- a/runners/agent-runner/package.json +++ b/runners/agent-runner/package.json @@ -13,6 +13,7 @@ "@anthropic-ai/claude-agent-sdk": "^0.2.87", "@modelcontextprotocol/sdk": "^1.12.1", "cron-parser": "^5.0.0", + "ejclaw-runners-shared": "file:../shared", "zod": "^4.0.0" }, "devDependencies": { diff --git a/runners/agent-runner/src/reviewer-runtime.ts b/runners/agent-runner/src/reviewer-runtime.ts index 7d80d4f..a0d2ed2 100644 --- a/runners/agent-runner/src/reviewer-runtime.ts +++ b/runners/agent-runner/src/reviewer-runtime.ts @@ -1,47 +1,22 @@ 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'; +export { + assertReadonlyWorkspaceRepoConnectivity, + buildReviewerGitGuardEnv, + isReviewerRuntime, +} from 'ejclaw-runners-shared'; export type ClaudeReadonlySandboxMode = 'strict' | 'best-effort'; -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 { - if (process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE === '1') { - return false; - } - return roomRoleContext?.role === 'reviewer'; -} - export function isClaudeReadonlyReviewerRuntime( roomRoleContext?: RoomRoleContext, ): boolean { @@ -107,11 +82,13 @@ export function buildClaudeReadonlySandboxSettings( platform, ), ) { - const normalizedPaths = [...new Set( - protectedPaths - .filter((value): value is string => Boolean(value)) - .map((value) => path.resolve(value)), - )]; + const normalizedPaths = [ + ...new Set( + protectedPaths + .filter((value): value is string => Boolean(value)) + .map((value) => path.resolve(value)), + ), + ]; return { enabled: true, @@ -119,236 +96,11 @@ export function buildClaudeReadonlySandboxSettings( autoAllowBashIfSandboxed: true, allowUnsandboxedCommands: false, filesystem: - normalizedPaths.length > 0 - ? { denyWrite: normalizedPaths } - : undefined, + normalizedPaths.length > 0 ? { denyWrite: normalizedPaths } : undefined, }; } -function resolveGitBinary(baseEnv: NodeJS.ProcessEnv): string { - return execFileSync('bash', ['-lc', 'command -v git'], { - encoding: 'utf-8', - env: baseEnv, - stdio: ['ignore', 'pipe', 'pipe'], - }).trim(); -} - -function createExecutableWrapperDir(baseEnv: NodeJS.ProcessEnv): string { - const candidateRoots = [ - baseEnv.EJCLAW_REVIEWER_GIT_WRAPPER_ROOT, - baseEnv.HOME ? path.join(baseEnv.HOME, '.ejclaw-reviewer-runtime') : null, - path.join(os.tmpdir(), '.ejclaw-reviewer-runtime'), - ].filter((value): value is string => Boolean(value)); - - const probeContents = '#!/usr/bin/env bash\nexit 0\n'; - const tried = new Set(); - - for (const candidateRoot of candidateRoots) { - if (tried.has(candidateRoot)) { - continue; - } - tried.add(candidateRoot); - try { - fs.mkdirSync(candidateRoot, { recursive: true }); - const wrapperDir = fs.mkdtempSync( - path.join(candidateRoot, 'ejclaw-reviewer-git-'), - ); - const probePath = path.join(wrapperDir, 'probe'); - fs.writeFileSync(probePath, probeContents, { mode: 0o755 }); - execFileSync(probePath, [], { - stdio: ['ignore', 'ignore', 'ignore'], - }); - fs.rmSync(probePath, { force: true }); - return wrapperDir; - } catch { - continue; - } - } - - throw new Error( - 'Unable to create an executable git guard wrapper directory for reviewer runtime.', - ); -} - -export function buildReviewerGitGuardEnv( - baseEnv: NodeJS.ProcessEnv, - reviewerRuntime: boolean, -): NodeJS.ProcessEnv { - if (!reviewerRuntime) { - return baseEnv; - } - - const realGitPath = resolveGitBinary(baseEnv); - const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; - const wrapperDir = createExecutableWrapperDir(baseEnv); - 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)} -protected_work_dir=${JSON.stringify(protectedWorkDir)} -blocked_subcommands=(${blocked}) -subcmd="" -skip_next=0 -target_dir="$(pwd -P)" -capture_next_dir=0 -for arg in "$@"; do - if [[ "$capture_next_dir" == "1" ]]; then - if [[ "$arg" == /* ]]; then - target_dir="$arg" - else - target_dir="$target_dir/$arg" - fi - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - capture_next_dir=0 - continue - fi - if [[ "$skip_next" == "1" ]]; then - skip_next=0 - continue - fi - case "$arg" in - -C) - capture_next_dir=1 - continue - ;; - -C*) - target_dir="\${arg#-C}" - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - continue - ;; - --work-tree) - capture_next_dir=1 - continue - ;; - --work-tree=*) - target_dir="\${arg#--work-tree=}" - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - continue - ;; - -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 -is_protected_target=0 -if [[ -n "$protected_work_dir" ]]; then - protected_real="$(cd "$protected_work_dir" 2>/dev/null && pwd -P || printf '%s' "$protected_work_dir")" - case "$target_dir" in - "$protected_real"|"$protected_real"/*) - is_protected_target=1 - ;; - esac -fi -for blocked in "\${blocked_subcommands[@]}"; do - if [[ "$is_protected_target" == "1" && "$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, - EJCLAW_PROTECTED_WORK_DIR: protectedWorkDir, - PATH: `${wrapperDir}:${baseEnv.PATH || ''}`, - }; -} - -function readGitOutput( - args: string[], - baseEnv: NodeJS.ProcessEnv, - cwd: string, -): string { - return execFileSync('git', args, { - cwd, - env: baseEnv, - encoding: 'utf-8', - stdio: ['ignore', 'pipe', 'pipe'], - }).trim(); -} - -function isRemoteGitOrigin(originUrl: string): boolean { - return ( - /^(?:https?|ssh|git):\/\//i.test(originUrl) || - /^[^/\\]+@[^:]+:.+/.test(originUrl) - ); -} - -export function assertReadonlyWorkspaceRepoConnectivity( - baseEnv: NodeJS.ProcessEnv, - enabled: boolean, -): void { - if (!enabled) { - return; - } - - const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; - if (!protectedWorkDir) { - return; - } - - let originUrl = ''; - try { - originUrl = readGitOutput( - ['config', '--get', 'remote.origin.url'], - baseEnv, - protectedWorkDir, - ); - } catch { - return; - } - - if (isRemoteGitOrigin(originUrl)) { - return; - } - - if (!path.isAbsolute(originUrl) || !fs.existsSync(originUrl)) { - throw new Error( - `EJClaw readonly runtime cannot access local git origin path: ${originUrl || '(missing)'}`, - ); - } - - try { - readGitOutput(['rev-parse', '--git-dir'], baseEnv, originUrl); - } catch { - throw new Error( - `EJClaw readonly runtime origin path is not mounted as a git repository: ${originUrl}`, - ); - } - - try { - readGitOutput(['ls-remote', 'origin', 'HEAD'], baseEnv, protectedWorkDir); - } catch { - throw new Error( - `EJClaw readonly runtime cannot resolve local git origin from ${protectedWorkDir}. Check canonical repo mount for ${originUrl}.`, - ); - } -} - export function isReviewerMutatingShellCommand(command: string): boolean { const normalized = command.trim(); - return ( - MUTATING_SHELL_PATTERNS.some((pattern) => pattern.test(normalized)) - ); + return MUTATING_SHELL_PATTERNS.some((pattern) => pattern.test(normalized)); } diff --git a/runners/agent-runner/src/room-role-context.ts b/runners/agent-runner/src/room-role-context.ts index 19d9381..588ab4e 100644 --- a/runners/agent-runner/src/room-role-context.ts +++ b/runners/agent-runner/src/room-role-context.ts @@ -1,23 +1,4 @@ -export interface RoomRoleContext { - serviceId: string; - role: 'owner' | 'reviewer'; - ownerServiceId: string; - reviewerServiceId: string; - failoverOwner: boolean; -} - -export function prependRoomRoleHeader( - prompt: string, - roomRoleContext?: RoomRoleContext, -): string { - if (!roomRoleContext) { - return prompt; - } - - const header = - `[ROOM_ROLE self=${roomRoleContext.serviceId} role=${roomRoleContext.role} ` + - `owner=${roomRoleContext.ownerServiceId} reviewer=${roomRoleContext.reviewerServiceId} ` + - `failover=${roomRoleContext.failoverOwner ? 1 : 0}]`; - - return `${header}\n\n${prompt}`; -} +export { + prependRoomRoleHeader, + type RoomRoleContext, +} from 'ejclaw-runners-shared'; diff --git a/runners/codex-runner/bun.lock b/runners/codex-runner/bun.lock index 835ab8a..eb75307 100644 --- a/runners/codex-runner/bun.lock +++ b/runners/codex-runner/bun.lock @@ -6,6 +6,7 @@ "name": "ejclaw-codex-runner", "dependencies": { "@openai/codex": "^0.117.0", + "ejclaw-runners-shared": "file:../shared", }, "devDependencies": { "@types/node": "^22.10.7", @@ -30,6 +31,8 @@ "@types/node": ["@types/node@22.19.15", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-F0R/h2+dsy5wJAUe3tAU6oqa2qbWY5TpNfL/RGmo1y38hiyO1w3x2jPtt76wmuaJI4DQnOBu21cNXQ2STIUUWg=="], + "ejclaw-runners-shared": ["ejclaw-runners-shared@file:../shared", { "devDependencies": { "@types/node": "^22.10.7", "typescript": "^5.7.3" } }], + "typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="], "undici-types": ["undici-types@6.21.0", "", {}, "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ=="], diff --git a/runners/codex-runner/package.json b/runners/codex-runner/package.json index a6cbd48..83e407f 100644 --- a/runners/codex-runner/package.json +++ b/runners/codex-runner/package.json @@ -10,6 +10,7 @@ "start": "bun dist/index.js" }, "dependencies": { + "ejclaw-runners-shared": "file:../shared", "@openai/codex": "^0.117.0" }, "devDependencies": { diff --git a/runners/codex-runner/src/reviewer-runtime.ts b/runners/codex-runner/src/reviewer-runtime.ts index 2dd530a..e19efb5 100644 --- a/runners/codex-runner/src/reviewer-runtime.ts +++ b/runners/codex-runner/src/reviewer-runtime.ts @@ -1,259 +1,9 @@ -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'; +export { + assertReadonlyWorkspaceRepoConnectivity, + buildReviewerGitGuardEnv, + isReviewerRuntime, +} from 'ejclaw-runners-shared'; // Codex app-server does not expose a BashTool-style pre-use hook, so reviewer // mode can only hard-block mutating git via PATH interception here. Non-git // shell mutation commands remain a known gap when REVIEWER_AGENT_TYPE=codex. -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 { - if (process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE === '1') { - return false; - } - 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(); -} - -function createExecutableWrapperDir(baseEnv: NodeJS.ProcessEnv): string { - const candidateRoots = [ - baseEnv.EJCLAW_REVIEWER_GIT_WRAPPER_ROOT, - baseEnv.HOME ? path.join(baseEnv.HOME, '.ejclaw-reviewer-runtime') : null, - path.join(os.tmpdir(), '.ejclaw-reviewer-runtime'), - ].filter((value): value is string => Boolean(value)); - - const probeContents = '#!/usr/bin/env bash\nexit 0\n'; - const tried = new Set(); - - for (const candidateRoot of candidateRoots) { - if (tried.has(candidateRoot)) { - continue; - } - tried.add(candidateRoot); - try { - fs.mkdirSync(candidateRoot, { recursive: true }); - const wrapperDir = fs.mkdtempSync( - path.join(candidateRoot, 'ejclaw-reviewer-git-'), - ); - const probePath = path.join(wrapperDir, 'probe'); - fs.writeFileSync(probePath, probeContents, { mode: 0o755 }); - execFileSync(probePath, [], { - stdio: ['ignore', 'ignore', 'ignore'], - }); - fs.rmSync(probePath, { force: true }); - return wrapperDir; - } catch { - continue; - } - } - - throw new Error( - 'Unable to create an executable git guard wrapper directory for reviewer runtime.', - ); -} - -export function buildReviewerGitGuardEnv( - baseEnv: NodeJS.ProcessEnv, - reviewerRuntime: boolean, -): NodeJS.ProcessEnv { - if (!reviewerRuntime) { - return baseEnv; - } - - const realGitPath = resolveGitBinary(baseEnv); - const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; - const wrapperDir = createExecutableWrapperDir(baseEnv); - 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)} -protected_work_dir=${JSON.stringify(protectedWorkDir)} -blocked_subcommands=(${blocked}) -subcmd="" -skip_next=0 -target_dir="$(pwd -P)" -capture_next_dir=0 -for arg in "$@"; do - if [[ "$capture_next_dir" == "1" ]]; then - if [[ "$arg" == /* ]]; then - target_dir="$arg" - else - target_dir="$target_dir/$arg" - fi - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - capture_next_dir=0 - continue - fi - if [[ "$skip_next" == "1" ]]; then - skip_next=0 - continue - fi - case "$arg" in - -C) - capture_next_dir=1 - continue - ;; - -C*) - target_dir="\${arg#-C}" - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - continue - ;; - --work-tree) - capture_next_dir=1 - continue - ;; - --work-tree=*) - target_dir="\${arg#--work-tree=}" - target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" - continue - ;; - -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 -is_protected_target=0 -if [[ -n "$protected_work_dir" ]]; then - protected_real="$(cd "$protected_work_dir" 2>/dev/null && pwd -P || printf '%s' "$protected_work_dir")" - case "$target_dir" in - "$protected_real"|"$protected_real"/*) - is_protected_target=1 - ;; - esac -fi -for blocked in "\${blocked_subcommands[@]}"; do - if [[ "$is_protected_target" == "1" && "$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, - EJCLAW_PROTECTED_WORK_DIR: protectedWorkDir, - PATH: `${wrapperDir}:${baseEnv.PATH || ''}`, - }; -} - -function readGitOutput( - args: string[], - baseEnv: NodeJS.ProcessEnv, - cwd: string, -): string { - return execFileSync('git', args, { - cwd, - env: baseEnv, - encoding: 'utf-8', - stdio: ['ignore', 'pipe', 'pipe'], - }).trim(); -} - -function isRemoteGitOrigin(originUrl: string): boolean { - return ( - /^(?:https?|ssh|git):\/\//i.test(originUrl) || - /^[^/\\]+@[^:]+:.+/.test(originUrl) - ); -} - -export function assertReadonlyWorkspaceRepoConnectivity( - baseEnv: NodeJS.ProcessEnv, - enabled: boolean, -): void { - if (!enabled) { - return; - } - - const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; - if (!protectedWorkDir) { - return; - } - - let originUrl = ''; - try { - originUrl = readGitOutput( - ['config', '--get', 'remote.origin.url'], - baseEnv, - protectedWorkDir, - ); - } catch { - return; - } - - if (isRemoteGitOrigin(originUrl)) { - return; - } - - if (!path.isAbsolute(originUrl) || !fs.existsSync(originUrl)) { - throw new Error( - `EJClaw readonly runtime cannot access local git origin path: ${originUrl || '(missing)'}`, - ); - } - - try { - readGitOutput(['rev-parse', '--git-dir'], baseEnv, originUrl); - } catch { - throw new Error( - `EJClaw readonly runtime origin path is not mounted as a git repository: ${originUrl}`, - ); - } - - try { - readGitOutput(['ls-remote', 'origin', 'HEAD'], baseEnv, protectedWorkDir); - } catch { - throw new Error( - `EJClaw readonly runtime cannot resolve local git origin from ${protectedWorkDir}. Check canonical repo mount for ${originUrl}.`, - ); - } -} diff --git a/runners/codex-runner/src/room-role-context.ts b/runners/codex-runner/src/room-role-context.ts index 19d9381..588ab4e 100644 --- a/runners/codex-runner/src/room-role-context.ts +++ b/runners/codex-runner/src/room-role-context.ts @@ -1,23 +1,4 @@ -export interface RoomRoleContext { - serviceId: string; - role: 'owner' | 'reviewer'; - ownerServiceId: string; - reviewerServiceId: string; - failoverOwner: boolean; -} - -export function prependRoomRoleHeader( - prompt: string, - roomRoleContext?: RoomRoleContext, -): string { - if (!roomRoleContext) { - return prompt; - } - - const header = - `[ROOM_ROLE self=${roomRoleContext.serviceId} role=${roomRoleContext.role} ` + - `owner=${roomRoleContext.ownerServiceId} reviewer=${roomRoleContext.reviewerServiceId} ` + - `failover=${roomRoleContext.failoverOwner ? 1 : 0}]`; - - return `${header}\n\n${prompt}`; -} +export { + prependRoomRoleHeader, + type RoomRoleContext, +} from 'ejclaw-runners-shared'; diff --git a/runners/shared/bun.lock b/runners/shared/bun.lock new file mode 100644 index 0000000..2db2714 --- /dev/null +++ b/runners/shared/bun.lock @@ -0,0 +1,20 @@ +{ + "lockfileVersion": 1, + "configVersion": 1, + "workspaces": { + "": { + "name": "ejclaw-runners-shared", + "devDependencies": { + "@types/node": "^22.10.7", + "typescript": "^5.7.3", + }, + }, + }, + "packages": { + "@types/node": ["@types/node@22.19.17", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-wGdMcf+vPYM6jikpS/qhg6WiqSV/OhG+jeeHT/KlVqxYfD40iYJf9/AE1uQxVWFvU7MipKRkRv8NSHiCGgPr8Q=="], + + "typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="], + + "undici-types": ["undici-types@6.21.0", "", {}, "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ=="], + } +} diff --git a/runners/shared/package.json b/runners/shared/package.json new file mode 100644 index 0000000..51ac14e --- /dev/null +++ b/runners/shared/package.json @@ -0,0 +1,15 @@ +{ + "name": "ejclaw-runners-shared", + "version": "1.0.0", + "type": "module", + "description": "Shared runner utilities for EJClaw", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "scripts": { + "build": "tsc" + }, + "devDependencies": { + "@types/node": "^22.10.7", + "typescript": "^5.7.3" + } +} diff --git a/runners/shared/src/index.ts b/runners/shared/src/index.ts new file mode 100644 index 0000000..4068126 --- /dev/null +++ b/runners/shared/src/index.ts @@ -0,0 +1,9 @@ +export { + prependRoomRoleHeader, + type RoomRoleContext, +} from './room-role-context.js'; +export { + assertReadonlyWorkspaceRepoConnectivity, + buildReviewerGitGuardEnv, + isReviewerRuntime, +} from './reviewer-git-guard.js'; diff --git a/runners/shared/src/reviewer-git-guard.ts b/runners/shared/src/reviewer-git-guard.ts new file mode 100644 index 0000000..3ee6432 --- /dev/null +++ b/runners/shared/src/reviewer-git-guard.ts @@ -0,0 +1,254 @@ +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 { + if (process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE === '1') { + return false; + } + 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(); +} + +function createExecutableWrapperDir(baseEnv: NodeJS.ProcessEnv): string { + const candidateRoots = [ + baseEnv.EJCLAW_REVIEWER_GIT_WRAPPER_ROOT, + baseEnv.HOME ? path.join(baseEnv.HOME, '.ejclaw-reviewer-runtime') : null, + path.join(os.tmpdir(), '.ejclaw-reviewer-runtime'), + ].filter((value): value is string => Boolean(value)); + + const probeContents = '#!/usr/bin/env bash\nexit 0\n'; + const tried = new Set(); + + for (const candidateRoot of candidateRoots) { + if (tried.has(candidateRoot)) { + continue; + } + tried.add(candidateRoot); + try { + fs.mkdirSync(candidateRoot, { recursive: true }); + const wrapperDir = fs.mkdtempSync( + path.join(candidateRoot, 'ejclaw-reviewer-git-'), + ); + const probePath = path.join(wrapperDir, 'probe'); + fs.writeFileSync(probePath, probeContents, { mode: 0o755 }); + execFileSync(probePath, [], { + stdio: ['ignore', 'ignore', 'ignore'], + }); + fs.rmSync(probePath, { force: true }); + return wrapperDir; + } catch { + continue; + } + } + + throw new Error( + 'Unable to create an executable git guard wrapper directory for reviewer runtime.', + ); +} + +export function buildReviewerGitGuardEnv( + baseEnv: NodeJS.ProcessEnv, + reviewerRuntime: boolean, +): NodeJS.ProcessEnv { + if (!reviewerRuntime) { + return baseEnv; + } + + const realGitPath = resolveGitBinary(baseEnv); + const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; + const wrapperDir = createExecutableWrapperDir(baseEnv); + 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)} +protected_work_dir=${JSON.stringify(protectedWorkDir)} +blocked_subcommands=(${blocked}) +subcmd="" +skip_next=0 +target_dir="$(pwd -P)" +capture_next_dir=0 +for arg in "$@"; do + if [[ "$capture_next_dir" == "1" ]]; then + if [[ "$arg" == /* ]]; then + target_dir="$arg" + else + target_dir="$target_dir/$arg" + fi + target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" + capture_next_dir=0 + continue + fi + if [[ "$skip_next" == "1" ]]; then + skip_next=0 + continue + fi + case "$arg" in + -C) + capture_next_dir=1 + continue + ;; + -C*) + target_dir="\${arg#-C}" + target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" + continue + ;; + --work-tree) + capture_next_dir=1 + continue + ;; + --work-tree=*) + target_dir="\${arg#--work-tree=}" + target_dir="$(cd "$target_dir" 2>/dev/null && pwd -P || printf '%s' "$target_dir")" + continue + ;; + -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 +is_protected_target=0 +if [[ -n "$protected_work_dir" ]]; then + protected_real="$(cd "$protected_work_dir" 2>/dev/null && pwd -P || printf '%s' "$protected_work_dir")" + case "$target_dir" in + "$protected_real"|"$protected_real"/*) + is_protected_target=1 + ;; + esac +fi +for blocked in "\${blocked_subcommands[@]}"; do + if [[ "$is_protected_target" == "1" && "$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, + EJCLAW_PROTECTED_WORK_DIR: protectedWorkDir, + PATH: `${wrapperDir}:${baseEnv.PATH || ''}`, + }; +} + +function readGitOutput( + args: string[], + baseEnv: NodeJS.ProcessEnv, + cwd: string, +): string { + return execFileSync('git', args, { + cwd, + env: baseEnv, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +function isRemoteGitOrigin(originUrl: string): boolean { + return ( + /^(?:https?|ssh|git):\/\//i.test(originUrl) || + /^[^/\\]+@[^:]+:.+/.test(originUrl) + ); +} + +export function assertReadonlyWorkspaceRepoConnectivity( + baseEnv: NodeJS.ProcessEnv, + enabled: boolean, +): void { + if (!enabled) { + return; + } + + const protectedWorkDir = baseEnv.EJCLAW_WORK_DIR || ''; + if (!protectedWorkDir) { + return; + } + + let originUrl = ''; + try { + originUrl = readGitOutput( + ['config', '--get', 'remote.origin.url'], + baseEnv, + protectedWorkDir, + ); + } catch { + return; + } + + if (isRemoteGitOrigin(originUrl)) { + return; + } + + if (!path.isAbsolute(originUrl) || !fs.existsSync(originUrl)) { + throw new Error( + `EJClaw readonly runtime cannot access local git origin path: ${originUrl || '(missing)'}`, + ); + } + + try { + readGitOutput(['rev-parse', '--git-dir'], baseEnv, originUrl); + } catch { + throw new Error( + `EJClaw readonly runtime origin path is not mounted as a git repository: ${originUrl}`, + ); + } + + try { + readGitOutput(['ls-remote', 'origin', 'HEAD'], baseEnv, protectedWorkDir); + } catch { + throw new Error( + `EJClaw readonly runtime cannot resolve local git origin from ${protectedWorkDir}. Check canonical repo mount for ${originUrl}.`, + ); + } +} diff --git a/runners/shared/src/room-role-context.ts b/runners/shared/src/room-role-context.ts new file mode 100644 index 0000000..19d9381 --- /dev/null +++ b/runners/shared/src/room-role-context.ts @@ -0,0 +1,23 @@ +export interface RoomRoleContext { + serviceId: string; + role: 'owner' | 'reviewer'; + ownerServiceId: string; + reviewerServiceId: string; + failoverOwner: boolean; +} + +export function prependRoomRoleHeader( + prompt: string, + roomRoleContext?: RoomRoleContext, +): string { + if (!roomRoleContext) { + return prompt; + } + + const header = + `[ROOM_ROLE self=${roomRoleContext.serviceId} role=${roomRoleContext.role} ` + + `owner=${roomRoleContext.ownerServiceId} reviewer=${roomRoleContext.reviewerServiceId} ` + + `failover=${roomRoleContext.failoverOwner ? 1 : 0}]`; + + return `${header}\n\n${prompt}`; +} diff --git a/runners/shared/tsconfig.json b/runners/shared/tsconfig.json new file mode 100644 index 0000000..de6431e --- /dev/null +++ b/runners/shared/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "outDir": "./dist", + "rootDir": "./src", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "declaration": true + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/src/agent-runner-process.ts b/src/agent-runner-process.ts new file mode 100644 index 0000000..37f7801 --- /dev/null +++ b/src/agent-runner-process.ts @@ -0,0 +1,428 @@ +import type { ChildProcess } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +import { + AGENT_MAX_OUTPUT_SIZE, + AGENT_TIMEOUT, + IDLE_TIMEOUT, + LOG_LEVEL, +} from './config.js'; +import { logger } from './logger.js'; +import { OUTPUT_END_MARKER, OUTPUT_START_MARKER } from './agent-protocol.js'; +import type { AgentInput, AgentOutput } from './agent-runner.js'; +import type { RegisteredGroup } from './types.js'; +import { getErrorMessage } from './utils.js'; + +interface RunSpawnedAgentProcessArgs { + proc: ChildProcess; + group: RegisteredGroup; + input: AgentInput; + processName: string; + logsDir: string; + startTime: number; + onOutput?: (output: AgentOutput) => Promise; +} + +function parseLegacyAgentOutput(stdout: string): AgentOutput { + const startIdx = stdout.indexOf(OUTPUT_START_MARKER); + const endIdx = stdout.indexOf(OUTPUT_END_MARKER); + let jsonLine: string; + if (startIdx !== -1 && endIdx !== -1 && endIdx > startIdx) { + jsonLine = stdout + .slice(startIdx + OUTPUT_START_MARKER.length, endIdx) + .trim(); + } else { + const lines = stdout.trim().split('\n'); + jsonLine = lines[lines.length - 1]; + } + return JSON.parse(jsonLine) as AgentOutput; +} + +export function runSpawnedAgentProcess( + args: RunSpawnedAgentProcessArgs, +): Promise { + const { proc, group, input, processName, logsDir, startTime, onOutput } = + args; + return new Promise((resolve) => { + const stdoutStream = proc.stdout; + const stderrStream = proc.stderr; + if (!stdoutStream || !stderrStream) { + resolve({ + status: 'error', + result: null, + error: 'Agent process stdio pipes are unavailable', + }); + return; + } + + let stdout = ''; + let stderr = ''; + let stdoutTruncated = false; + let stderrTruncated = false; + + // Streaming output: parse OUTPUT_START/END marker pairs as they arrive. + let parseBuffer = ''; + let newSessionId: string | undefined; + let outputChain = Promise.resolve(); + + let timedOut = false; + let hadStreamingOutput = false; + const configTimeout = group.agentConfig?.timeout || AGENT_TIMEOUT; + const timeoutMs = Math.max(configTimeout, IDLE_TIMEOUT + 30_000); + + const killOnTimeout = () => { + timedOut = true; + logger.error( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + }, + 'Agent timeout, sending SIGTERM', + ); + proc.kill('SIGTERM'); + setTimeout(() => { + if (!proc.killed) proc.kill('SIGKILL'); + }, 15000); + }; + + let timeout = setTimeout(killOnTimeout, timeoutMs); + + const resetTimeout = () => { + clearTimeout(timeout); + timeout = setTimeout(killOnTimeout, timeoutMs); + }; + + stdoutStream.on('data', (data) => { + const chunk = data.toString(); + + if (!stdoutTruncated) { + const remaining = AGENT_MAX_OUTPUT_SIZE - stdout.length; + if (chunk.length > remaining) { + stdout += chunk.slice(0, remaining); + stdoutTruncated = true; + logger.warn( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + size: stdout.length, + }, + 'Agent stdout truncated due to size limit', + ); + } else { + stdout += chunk; + } + } + + if (!onOutput) { + return; + } + + parseBuffer += chunk; + let startIdx: number; + while ((startIdx = parseBuffer.indexOf(OUTPUT_START_MARKER)) !== -1) { + const endIdx = parseBuffer.indexOf(OUTPUT_END_MARKER, startIdx); + if (endIdx === -1) break; + + const jsonStr = parseBuffer + .slice(startIdx + OUTPUT_START_MARKER.length, endIdx) + .trim(); + parseBuffer = parseBuffer.slice(endIdx + OUTPUT_END_MARKER.length); + + try { + const parsed: AgentOutput = JSON.parse(jsonStr); + if (parsed.newSessionId) { + newSessionId = parsed.newSessionId; + } + hadStreamingOutput = true; + resetTimeout(); + if (parsed.status === 'error') { + logger.warn( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: parsed.error, + newSessionId: parsed.newSessionId, + }, + 'Streamed agent error output', + ); + } + outputChain = outputChain.then(() => onOutput(parsed)); + } catch (err) { + logger.warn( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: err, + }, + 'Failed to parse streamed output chunk', + ); + } + } + }); + + stderrStream.on('data', (data) => { + const chunk = data.toString(); + const lines = chunk.trim().split('\n'); + for (const line of lines) { + if (!line) continue; + if ( + line.includes('Turn in progress') || + line.includes('Subagent') || + line.includes('Intermediate assistant') || + line.includes('Promoting') || + line.includes('Flushing') || + line.includes('Result #') || + line.includes('Query done') || + line.includes('Terminal') || + line.includes('Assistant: stop=') || + line.includes('Close sentinel') + ) { + logger.info( + { group: group.name, chatJid: input.chatJid, runId: input.runId }, + line.replace(/^\[.*?\]\s*/, ''), + ); + } else { + logger.debug( + { agent: group.folder, chatJid: input.chatJid, runId: input.runId }, + line, + ); + } + } + resetTimeout(); + if (stderrTruncated) return; + const remaining = AGENT_MAX_OUTPUT_SIZE - stderr.length; + if (chunk.length > remaining) { + stderr += chunk.slice(0, remaining); + stderrTruncated = true; + } else { + stderr += chunk; + } + }); + + proc.on('close', (code, signal) => { + clearTimeout(timeout); + const duration = Date.now() - startTime; + + if (timedOut) { + const ts = new Date().toISOString().replace(/[:.]/g, '-'); + fs.writeFileSync( + path.join(logsDir, `agent-${input.runId || 'adhoc'}-${ts}.log`), + [ + `=== Agent Run Log (TIMEOUT) ===`, + `Timestamp: ${new Date().toISOString()}`, + `Group: ${group.name}`, + `ChatJid: ${input.chatJid}`, + `RunId: ${input.runId || 'n/a'}`, + `Process: ${processName}`, + `Duration: ${duration}ms`, + `Exit Code: ${code}`, + `Signal: ${signal}`, + `Had Streaming Output: ${hadStreamingOutput}`, + ].join('\n'), + ); + + if (hadStreamingOutput) { + logger.info( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + duration, + code, + signal, + }, + 'Agent timed out after output (idle cleanup)', + ); + outputChain.then(() => { + resolve({ status: 'success', result: null, newSessionId }); + }); + return; + } + + resolve({ + status: 'error', + result: null, + error: `Agent timed out after ${configTimeout}ms`, + }); + return; + } + + if (code === null && signal) { + if (hadStreamingOutput) { + logger.info( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + duration, + signal, + }, + 'Agent terminated by signal after output delivery (normal cleanup)', + ); + outputChain.then(() => { + resolve({ status: 'success', result: null, newSessionId }); + }); + return; + } + + logger.error( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + duration, + signal, + }, + 'Agent killed by signal before producing output', + ); + outputChain.then(() => { + resolve({ + status: 'error', + result: null, + error: `Agent killed by ${signal} before producing output`, + }); + }); + return; + } + + const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); + const logFile = path.join( + logsDir, + `agent-${input.runId || 'adhoc'}-${timestamp}.log`, + ); + const isVerbose = LOG_LEVEL === 'debug' || LOG_LEVEL === 'trace'; + const logLines = [ + `=== Agent Run Log ===`, + `Timestamp: ${new Date().toISOString()}`, + `Group: ${group.name}`, + `ChatJid: ${input.chatJid}`, + `GroupFolder: ${input.groupFolder}`, + `RunId: ${input.runId || 'n/a'}`, + `IsMain: ${input.isMain}`, + `AgentType: ${group.agentType || 'claude-code'}`, + `Duration: ${duration}ms`, + `Exit Code: ${code}`, + `Signal: ${signal}`, + ``, + ]; + + const isError = code !== 0; + if (isVerbose || isError) { + logLines.push( + `=== Input ===`, + JSON.stringify(input, null, 2), + ``, + `=== Stderr ===`, + stderr, + ``, + `=== Stdout ===`, + stdout, + ); + } else { + logLines.push( + `Prompt length: ${input.prompt.length} chars`, + `Session ID: ${input.sessionId || 'new'}`, + ); + } + + fs.writeFileSync(logFile, logLines.join('\n')); + + if (code !== 0) { + logger.error( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + code, + duration, + logFile, + }, + 'Agent exited with error', + ); + outputChain.then(() => { + resolve({ + status: 'error', + result: null, + error: `Agent exited with code ${code}: ${stderr.slice(-200)}`, + }); + }); + return; + } + + if (onOutput) { + outputChain.then(() => { + logger.info( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + duration, + newSessionId, + }, + 'Agent completed (streaming mode)', + ); + resolve({ status: 'success', result: null, newSessionId }); + }); + return; + } + + try { + const output = parseLegacyAgentOutput(stdout); + logger.info( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + duration, + status: output.status, + }, + 'Agent completed', + ); + resolve(output); + } catch (err) { + logger.error( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: err, + }, + 'Failed to parse agent output', + ); + resolve({ + status: 'error', + result: null, + error: `Failed to parse agent output: ${getErrorMessage(err)}`, + }); + } + }); + + proc.on('error', (err) => { + clearTimeout(timeout); + logger.error( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + error: err, + }, + 'Agent spawn error', + ); + resolve({ + status: 'error', + result: null, + error: `Agent spawn error: ${err.message}`, + }); + }); + }); +} diff --git a/src/agent-runner.ts b/src/agent-runner.ts index b5114f0..98b1806 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -1,22 +1,16 @@ -/** - * Agent Process Runner for EJClaw - * Spawns agent execution as direct host processes and handles IPC - */ import { ChildProcess, spawn } from 'child_process'; import fs from 'fs'; import path from 'path'; -import { getErrorMessage } from './utils.js'; -import { - AGENT_MAX_OUTPUT_SIZE, - AGENT_TIMEOUT, - IDLE_TIMEOUT, - LOG_LEVEL, -} from './config.js'; +/** + * Agent Process Runner for EJClaw + * Spawns agent execution as direct host processes and handles IPC. + */ import { prepareReadonlySessionEnvironment, prepareGroupEnvironment, } from './agent-runner-environment.js'; +import { runSpawnedAgentProcess } from './agent-runner-process.js'; import { getEnv } from './env.js'; export { type AvailableGroup, @@ -24,14 +18,7 @@ export { writeTasksSnapshot, } from './agent-runner-snapshot.js'; import { logger } from './logger.js'; -import { OUTPUT_END_MARKER, OUTPUT_START_MARKER } from './agent-protocol.js'; -import { - AgentOutputPhase, - AgentType, - RegisteredGroup, - RoomRoleContext, - StructuredAgentOutput, -} from './types.js'; +import { AgentType, RegisteredGroup, RoomRoleContext } from './types.js'; export interface AgentInput { prompt: string; @@ -52,8 +39,8 @@ export interface AgentInput { export interface AgentOutput { status: 'success' | 'error'; result: string | null; - output?: StructuredAgentOutput; - phase?: AgentOutputPhase; + output?: import('./types.js').StructuredAgentOutput; + phase?: import('./types.js').AgentOutputPhase; agentId?: string; agentLabel?: string; agentDone?: boolean; @@ -163,395 +150,17 @@ export async function runAgentProcess( onProcess(proc, processName, env.EJCLAW_IPC_DIR); - let stdout = ''; - let stderr = ''; - let stdoutTruncated = false; - let stderrTruncated = false; - proc.stdin.write(JSON.stringify(input)); proc.stdin.end(); - // Streaming output: parse OUTPUT_START/END marker pairs as they arrive - let parseBuffer = ''; - let newSessionId: string | undefined; - let outputChain = Promise.resolve(); - - proc.stdout.on('data', (data) => { - const chunk = data.toString(); - - if (!stdoutTruncated) { - const remaining = AGENT_MAX_OUTPUT_SIZE - stdout.length; - if (chunk.length > remaining) { - stdout += chunk.slice(0, remaining); - stdoutTruncated = true; - logger.warn( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - size: stdout.length, - }, - 'Agent stdout truncated due to size limit', - ); - } else { - stdout += chunk; - } - } - - if (onOutput) { - parseBuffer += chunk; - let startIdx: number; - while ((startIdx = parseBuffer.indexOf(OUTPUT_START_MARKER)) !== -1) { - const endIdx = parseBuffer.indexOf(OUTPUT_END_MARKER, startIdx); - if (endIdx === -1) break; - - const jsonStr = parseBuffer - .slice(startIdx + OUTPUT_START_MARKER.length, endIdx) - .trim(); - parseBuffer = parseBuffer.slice(endIdx + OUTPUT_END_MARKER.length); - - try { - const parsed: AgentOutput = JSON.parse(jsonStr); - if (parsed.newSessionId) { - newSessionId = parsed.newSessionId; - } - hadStreamingOutput = true; - resetTimeout(); - if (parsed.status === 'error') { - logger.warn( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - error: parsed.error, - newSessionId: parsed.newSessionId, - }, - 'Streamed agent error output', - ); - } - outputChain = outputChain.then(() => onOutput(parsed)); - } catch (err) { - logger.warn( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - error: err, - }, - 'Failed to parse streamed output chunk', - ); - } - } - } - }); - - let timedOut = false; - let hadStreamingOutput = false; - const configTimeout = group.agentConfig?.timeout || AGENT_TIMEOUT; - const timeoutMs = Math.max(configTimeout, IDLE_TIMEOUT + 30_000); - - const killOnTimeout = () => { - timedOut = true; - logger.error( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - processName, - }, - 'Agent timeout, sending SIGTERM', - ); - proc.kill('SIGTERM'); - // Force kill after 15s if still alive - setTimeout(() => { - if (!proc.killed) proc.kill('SIGKILL'); - }, 15000); - }; - - let timeout = setTimeout(killOnTimeout, timeoutMs); - - const resetTimeout = () => { - clearTimeout(timeout); - timeout = setTimeout(killOnTimeout, timeoutMs); - }; - - proc.stderr.on('data', (data) => { - const chunk = data.toString(); - const lines = chunk.trim().split('\n'); - for (const line of lines) { - if (!line) continue; - if ( - line.includes('Turn in progress') || - line.includes('Subagent') || - line.includes('Intermediate assistant') || - line.includes('Promoting') || - line.includes('Flushing') || - line.includes('Result #') || - line.includes('Query done') || - line.includes('Terminal') || - line.includes('Assistant: stop=') || - line.includes('Close sentinel') - ) { - logger.info( - { group: group.name, chatJid: input.chatJid, runId: input.runId }, - line.replace(/^\[.*?\]\s*/, ''), - ); - } else { - logger.debug( - { agent: group.folder, chatJid: input.chatJid, runId: input.runId }, - line, - ); - } - } - // Stderr activity means agent is alive — reset timeout - resetTimeout(); - if (stderrTruncated) return; - const remaining = AGENT_MAX_OUTPUT_SIZE - stderr.length; - if (chunk.length > remaining) { - stderr += chunk.slice(0, remaining); - stderrTruncated = true; - } else { - stderr += chunk; - } - }); - - proc.on('close', (code, signal) => { - clearTimeout(timeout); - const duration = Date.now() - startTime; - - if (timedOut) { - const ts = new Date().toISOString().replace(/[:.]/g, '-'); - fs.writeFileSync( - path.join(logsDir, `agent-${input.runId || 'adhoc'}-${ts}.log`), - [ - `=== Agent Run Log (TIMEOUT) ===`, - `Timestamp: ${new Date().toISOString()}`, - `Group: ${group.name}`, - `ChatJid: ${input.chatJid}`, - `RunId: ${input.runId || 'n/a'}`, - `Process: ${processName}`, - `Duration: ${duration}ms`, - `Exit Code: ${code}`, - `Signal: ${signal}`, - `Had Streaming Output: ${hadStreamingOutput}`, - ].join('\n'), - ); - - if (hadStreamingOutput) { - logger.info( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - processName, - duration, - code, - signal, - }, - 'Agent timed out after output (idle cleanup)', - ); - outputChain.then(() => { - resolve({ status: 'success', result: null, newSessionId }); - }); - return; - } - - resolve({ - status: 'error', - result: null, - error: `Agent timed out after ${configTimeout}ms`, - }); - return; - } - - // Signal kill (SIGTERM/SIGKILL) from post-close cleanup or service - // restart. When the agent already delivered streaming output this is - // normal lifecycle — not an error. - if (code === null && signal) { - if (hadStreamingOutput) { - logger.info( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - processName, - duration, - signal, - }, - 'Agent terminated by signal after output delivery (normal cleanup)', - ); - outputChain.then(() => { - resolve({ status: 'success', result: null, newSessionId }); - }); - return; - } - // No output delivered before signal kill — genuine error - logger.error( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - processName, - duration, - signal, - }, - 'Agent killed by signal before producing output', - ); - outputChain.then(() => { - resolve({ - status: 'error', - result: null, - error: `Agent killed by ${signal} before producing output`, - }); - }); - return; - } - - const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); - const logFile = path.join( - logsDir, - `agent-${input.runId || 'adhoc'}-${timestamp}.log`, - ); - const isVerbose = LOG_LEVEL === 'debug' || LOG_LEVEL === 'trace'; - - const logLines = [ - `=== Agent Run Log ===`, - `Timestamp: ${new Date().toISOString()}`, - `Group: ${group.name}`, - `ChatJid: ${input.chatJid}`, - `GroupFolder: ${input.groupFolder}`, - `RunId: ${input.runId || 'n/a'}`, - `IsMain: ${input.isMain}`, - `AgentType: ${group.agentType || 'claude-code'}`, - `Duration: ${duration}ms`, - `Exit Code: ${code}`, - `Signal: ${signal}`, - ``, - ]; - - const isError = code !== 0; - if (isVerbose || isError) { - logLines.push( - `=== Input ===`, - JSON.stringify(input, null, 2), - ``, - `=== Stderr ===`, - stderr, - ``, - `=== Stdout ===`, - stdout, - ); - } else { - logLines.push( - `Prompt length: ${input.prompt.length} chars`, - `Session ID: ${input.sessionId || 'new'}`, - ); - } - - fs.writeFileSync(logFile, logLines.join('\n')); - - if (code !== 0) { - logger.error( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - code, - duration, - logFile, - }, - 'Agent exited with error', - ); - // Wait for any queued streamed-output handlers to finish so a late - // newSessionId cannot be persisted after the caller clears a poisoned - // session. - outputChain.then(() => { - resolve({ - status: 'error', - result: null, - error: `Agent exited with code ${code}: ${stderr.slice(-200)}`, - }); - }); - return; - } - - if (onOutput) { - outputChain.then(() => { - logger.info( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - duration, - newSessionId, - }, - 'Agent completed (streaming mode)', - ); - resolve({ status: 'success', result: null, newSessionId }); - }); - return; - } - - // Legacy mode: parse output from stdout - try { - const startIdx = stdout.indexOf(OUTPUT_START_MARKER); - const endIdx = stdout.indexOf(OUTPUT_END_MARKER); - let jsonLine: string; - if (startIdx !== -1 && endIdx !== -1 && endIdx > startIdx) { - jsonLine = stdout - .slice(startIdx + OUTPUT_START_MARKER.length, endIdx) - .trim(); - } else { - const lines = stdout.trim().split('\n'); - jsonLine = lines[lines.length - 1]; - } - const output: AgentOutput = JSON.parse(jsonLine); - logger.info( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - duration, - status: output.status, - }, - 'Agent completed', - ); - resolve(output); - } catch (err) { - logger.error( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - error: err, - }, - 'Failed to parse agent output', - ); - resolve({ - status: 'error', - result: null, - error: `Failed to parse agent output: ${getErrorMessage(err)}`, - }); - } - }); - - proc.on('error', (err) => { - clearTimeout(timeout); - logger.error( - { - group: group.name, - chatJid: input.chatJid, - runId: input.runId, - processName, - error: err, - }, - 'Agent spawn error', - ); - resolve({ - status: 'error', - result: null, - error: `Agent spawn error: ${err.message}`, - }); - }); + runSpawnedAgentProcess({ + proc, + group, + input, + processName, + logsDir, + startTime, + onOutput, + }).then(resolve); }); } diff --git a/src/db.ts b/src/db.ts index 01fe3e2..fee1555 100644 --- a/src/db.ts +++ b/src/db.ts @@ -7,53 +7,24 @@ import { CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID, CODEX_REVIEW_SERVICE_ID, - DATA_DIR, - normalizeServiceId, - OWNER_AGENT_TYPE, SERVICE_ID, SERVICE_SESSION_SCOPE, } from './config.js'; -import { listUnexpectedDataStateFiles } from './data-state-files.js'; import { - isValidGroupFolder, resolveTaskRuntimeIpcPath as resolveTaskRuntimeIpcPathFromGroup, resolveServiceTaskSessionsPath as resolveServiceTaskSessionsPathFromGroup, resolveTaskSessionsPath as resolveTaskSessionsPathFromGroup, } from './group-folder.js'; import { logger } from './logger.js'; import { - type RoomModeSource, - type RoomRegistrationSnapshot, - countPendingLegacyRegisteredGroupRows, type StoredRoomSettings, - buildRegisteredGroupFromStoredSettings, - buildLegacyRoomMigrationPlan, - collectRegisteredAgentTypes, - collectRegisteredAgentTypesForFolder, - deleteLegacyRegisteredGroupRowsForJid, - getPendingLegacyRegisteredGroupJids as getPendingLegacyRegisteredGroupJidsFromDatabase, - getStoredRoomRowsFromDatabase, - getStoredRoomSettingsRowFromDatabase, - inferStoredRoomCapabilityTypes, - inferOwnerAgentTypeFromRegisteredAgentTypes, inferRoomModeFromRegisteredAgentTypes, - insertStoredRoomSettings, - insertStoredRoomSettingsFromMigration, - materializeRegisteredGroupsForRoom, - normalizeRoomModeSource, - normalizeStoredAgentType, - resolveStoredRoomCapabilityTypes, - resolveAssignedRoomFolder, - syncRoomRoleOverridesForRoom, - upsertRoomRoleOverride, - updateStoredRoomMetadata, } from './db/room-registration.js'; import { - initializeDatabaseSchema, - openDatabaseFromFile, - openInMemoryDatabase, - openPersistentDatabase, -} from './db/bootstrap.js'; + openInitializedDatabaseFromFile, + openInitializedInMemoryDatabase, + openInitializedPersistentDatabase, +} from './db/database-lifecycle.js'; import { clearChannelOwnerLeaseInDatabase, getAllChannelOwnerLeasesFromDatabase, @@ -102,8 +73,6 @@ import { } from './db/work-items.js'; import { claimPairedTurnReservationInDatabase, - clearPairedTaskExecutionLeasesForServiceInDatabase, - clearExpiredPairedTaskExecutionLeasesInDatabase, clearPairedTaskExecutionLeasesInDatabase, clearPairedTurnReservationsInDatabase, type PairedTaskUpdates, @@ -156,12 +125,29 @@ import { } from './db/service-handoffs.js'; import { getLastRespondingAgentTypeFromDatabase, - getUnsupportedRouterStateKeysFromDatabase, getRouterStateForServiceFromDatabase, getRouterStateFromDatabase, setRouterStateForServiceInDatabase, setRouterStateInDatabase, } from './db/router-state.js'; +import { + type AssignRoomInput, + assignRoomInDatabase, + clearExplicitRoomModeInDatabase, + deleteStoredRoomSettingsForTestsInDatabase, + getAllRoomBindingsFromDatabase, + getEffectiveRoomModeFromDatabase, + getEffectiveRuntimeRoomModeFromDatabase, + getExplicitRoomModeFromDatabase, + getRegisteredAgentTypesForJidFromDatabase, + getRegisteredGroupFromDatabase, + getStoredRoomSettingsFromDatabase, + migrateLegacyRoomRegistrationsForTestsInDatabase, + setExplicitRoomModeInDatabase, + setRegisteredGroupForTestsInDatabase, + setStoredRoomOwnerAgentTypeForTestsInDatabase, + updateRegisteredGroupNameInDatabase, +} from './db/rooms.js'; import { deleteAllSessionsForGroupFromDatabase, deleteSessionFromDatabase, @@ -169,7 +155,6 @@ import { getSessionFromDatabase, setSessionInDatabase, } from './db/sessions.js'; -import { tableHasColumn } from './db/schema.js'; import { type CreateScheduledTaskInput, type ScheduledTaskStatusTrackingUpdates, @@ -187,10 +172,7 @@ import { updateTaskInDatabase, updateTaskStatusTrackingInDatabase, } from './db/tasks.js'; -import { - inferAgentTypeFromServiceShadow, - resolveRoleServiceShadow, -} from './role-service-shadow.js'; +import { inferAgentTypeFromServiceShadow } from './role-service-shadow.js'; import { getTaskRuntimeTaskId } from './task-watch-status.js'; import { NewMessage, @@ -210,22 +192,7 @@ import { export { inferRoomModeFromRegisteredAgentTypes }; let db: Database; - -interface StoredRoomModeRow { - roomMode: RoomMode; - source: RoomModeSource; -} - -export interface AssignRoomInput { - name: string; - roomMode?: RoomMode; - ownerAgentType?: AgentType; - folder?: string; - isMain?: boolean; - workDir?: string; - addedAt?: string; - ownerAgentConfig?: RegisteredGroup['agentConfig']; -} +export type { AssignRoomInput } from './db/rooms.js'; export type { MemoryRecord, MemoryScopeKind, @@ -240,40 +207,17 @@ export type { ChannelOwnerLeaseRow } from './db/channel-owner-leases.js'; export type { ServiceHandoff } from './db/service-handoffs.js'; export function initDatabase(): void { - db = openPersistentDatabase(); - initializeDatabaseSchema(db); - assertNoPendingLegacyRoomMigration(); - assertNoUnexpectedDataStateFiles(); - assertNoUnsupportedRouterStateDbKeys(); - clearPairedTaskExecutionLeasesForServiceInDatabase( - db, - normalizeServiceId(SERVICE_ID), - ); - clearExpiredPairedTaskExecutionLeasesInDatabase(db); + db = openInitializedPersistentDatabase(); } /** @internal - for tests only. Creates a fresh in-memory database. */ export function _initTestDatabase(): void { - db = openInMemoryDatabase(); - initializeDatabaseSchema(db); - clearPairedTaskExecutionLeasesForServiceInDatabase( - db, - normalizeServiceId(SERVICE_ID), - ); - clearExpiredPairedTaskExecutionLeasesInDatabase(db); + db = openInitializedInMemoryDatabase(); } /** @internal - for tests only. Opens an existing database file and runs schema/migrations. */ export function _initTestDatabaseFromFile(dbPath: string): void { - db = openDatabaseFromFile(dbPath); - initializeDatabaseSchema(db); - assertNoPendingLegacyRoomMigration(); - assertNoUnsupportedRouterStateDbKeys(); - clearPairedTaskExecutionLeasesForServiceInDatabase( - db, - normalizeServiceId(SERVICE_ID), - ); - clearExpiredPairedTaskExecutionLeasesInDatabase(db); + db = openInitializedDatabaseFromFile(dbPath); } /** @internal - for tests only. */ @@ -284,21 +228,7 @@ export function _setStoredRoomOwnerAgentTypeForTests( if (!db) { throw new Error('Database not initialized'); } - db.prepare( - `UPDATE room_settings - SET owner_agent_type = ?, - updated_at = ? - WHERE chat_jid = ?`, - ).run(ownerAgentType, new Date().toISOString(), chatJid); - if (ownerAgentType) { - const now = new Date().toISOString(); - upsertRoomRoleOverride(db, chatJid, { - role: 'owner', - agentType: ownerAgentType, - createdAt: now, - updatedAt: now, - }); - } + setStoredRoomOwnerAgentTypeForTestsInDatabase(db, chatJid, ownerAgentType); } /** @internal - for tests only. */ @@ -306,7 +236,7 @@ export function _deleteStoredRoomSettingsForTests(chatJid: string): void { if (!db) { throw new Error('Database not initialized'); } - db.prepare('DELETE FROM room_settings WHERE chat_jid = ?').run(chatJid); + deleteStoredRoomSettingsForTestsInDatabase(db, chatJid); } /** @internal - for tests only. */ @@ -745,84 +675,7 @@ export function getRegisteredGroup( jid: string, agentType?: string, ): (RegisteredGroup & { jid: string }) | undefined { - const requestedAgentType = normalizeStoredAgentType(agentType); - const stored = getStoredRoomSettingsRowFromDatabase(db, jid); - if (stored) { - return buildRegisteredGroupFromStoredSettings( - db, - stored, - requestedAgentType, - ); - } - return undefined; -} - -function writeLegacyRegisteredGroupAndSyncRoomSettings( - jid: string, - group: RegisteredGroup, -): void { - const existingStored = getStoredRoomSettingsRowFromDatabase(db, jid); - const existingRoomMode = getStoredRoomModeRow(jid); - if (!isValidGroupFolder(group.folder)) { - throw new Error(`Invalid group folder "${group.folder}" for JID ${jid}`); - } - const tx = db.transaction(() => { - const seededAgentType = group.agentType || 'claude-code'; - const agentTypes = new Set(collectRegisteredAgentTypes(db, jid)); - agentTypes.add(seededAgentType); - const inferredRoomMode = inferRoomModeFromRegisteredAgentTypes([ - ...agentTypes, - ]); - const roomMode = - existingRoomMode?.source === 'explicit' - ? existingRoomMode.roomMode - : inferredRoomMode; - const ownerAgentType = - existingStored?.modeSource === 'explicit' && existingStored.ownerAgentType - ? existingStored.ownerAgentType - : inferOwnerAgentTypeFromRegisteredAgentTypes([...agentTypes]); - const snapshot: RoomRegistrationSnapshot = { - name: group.name, - folder: group.folder, - triggerPattern: - existingStored?.modeSource === 'explicit' && existingStored.trigger - ? existingStored.trigger - : (group.trigger ?? ''), - requiresTrigger: - group.requiresTrigger ?? existingStored?.requiresTrigger ?? false, - isMain: group.isMain ?? existingStored?.isMain ?? false, - ownerAgentType, - workDir: group.workDir ?? existingStored?.workDir ?? null, - }; - - if (existingStored) { - updateStoredRoomMetadata(db, jid, snapshot); - if (!existingRoomMode || existingRoomMode.source === 'inferred') { - upsertStoredRoomMode(jid, roomMode, 'inferred'); - } - } else { - insertStoredRoomSettings(db, jid, roomMode, 'inferred', snapshot); - } - - materializeRegisteredGroupsForRoom( - db, - jid, - snapshot, - roomMode, - ownerAgentType, - seededAgentType === ownerAgentType ? group.agentConfig : undefined, - group.added_at, - ); - syncRoomRoleOverridesForRoom( - db, - jid, - roomMode, - ownerAgentType, - seededAgentType === ownerAgentType ? group.agentConfig : undefined, - group.added_at, - ); - }); - tx(); + return getRegisteredGroupFromDatabase(db, jid, agentType); } /** @@ -833,7 +686,7 @@ export function _setRegisteredGroupForTests( jid: string, group: RegisteredGroup, ): void { - writeLegacyRegisteredGroupAndSyncRoomSettings(jid, group); + setRegisteredGroupForTestsInDatabase(db, jid, group); } export function _migrateLegacyRoomRegistrationsForTests(): { @@ -843,294 +696,56 @@ export function _migrateLegacyRoomRegistrationsForTests(): { if (!db) { throw new Error('Database not initialized'); } - - const jids = getPendingLegacyRegisteredGroupJidsFromDatabase(db); - let migratedRooms = 0; - let migratedRoleOverrides = 0; - - db.transaction(() => { - for (const jid of jids) { - const plan = buildLegacyRoomMigrationPlan(db, jid); - if (!plan) continue; - insertStoredRoomSettingsFromMigration(db, plan); - for (const override of plan.roleOverrides) { - upsertRoomRoleOverride(db, jid, override); - migratedRoleOverrides += 1; - } - migratedRooms += 1; - } - })(); - - return { migratedRooms, migratedRoleOverrides }; + return migrateLegacyRoomRegistrationsForTestsInDatabase(db); } export function assignRoom( chatJid: string, input: AssignRoomInput, ): (RegisteredGroup & { jid: string }) | undefined { - const existing = getStoredRoomSettingsRowFromDatabase(db, chatJid); - const roomMode = input.roomMode || existing?.roomMode || 'single'; - const ownerAgentType = - input.ownerAgentType || existing?.ownerAgentType || OWNER_AGENT_TYPE; - const folder = resolveAssignedRoomFolder( - db, - chatJid, - input.name, - input.folder, - ); - const snapshot: RoomRegistrationSnapshot = { - name: input.name, - folder, - triggerPattern: existing?.trigger ?? '', - requiresTrigger: existing?.requiresTrigger ?? false, - isMain: input.isMain ?? existing?.isMain ?? false, - ownerAgentType, - workDir: input.workDir ?? existing?.workDir ?? null, - }; - const now = new Date().toISOString(); - - db.transaction(() => { - if (existing) { - db.prepare( - `UPDATE room_settings - SET room_mode = ?, - mode_source = 'explicit', - name = ?, - folder = ?, - trigger_pattern = ?, - requires_trigger = ?, - is_main = ?, - owner_agent_type = ?, - work_dir = ?, - updated_at = ? - WHERE chat_jid = ?`, - ).run( - roomMode, - snapshot.name, - snapshot.folder, - snapshot.triggerPattern, - snapshot.requiresTrigger ? 1 : 0, - snapshot.isMain ? 1 : 0, - snapshot.ownerAgentType, - snapshot.workDir, - now, - chatJid, - ); - } else { - insertStoredRoomSettings(db, chatJid, roomMode, 'explicit', snapshot); - } - - syncRoomRoleOverridesForRoom( - db, - chatJid, - roomMode, - ownerAgentType, - input.ownerAgentConfig, - input.addedAt ?? now, - ); - deleteLegacyRegisteredGroupRowsForJid(db, chatJid); - })(); - - return getRegisteredGroup(chatJid); + return assignRoomInDatabase(db, chatJid, input); } export function updateRegisteredGroupName(jid: string, name: string): void { - const plan = buildRoomRegistrationPlanForJid(jid, { name }); - if (!plan) { - return; - } - db.transaction(() => { - if (plan.hasStoredRoom) { - updateStoredRoomMetadata(db, jid, plan.snapshot); - } else { - insertStoredRoomSettings( - db, - jid, - plan.roomMode, - 'inferred', - plan.snapshot, - ); - } - deleteLegacyRegisteredGroupRowsForJid(db, jid); - })(); + updateRegisteredGroupNameInDatabase(db, jid, name); } export function getAllRoomBindings( agentTypeFilter?: string, ): Record { - const result: Record = {}; - const requestedAgentType = normalizeStoredAgentType(agentTypeFilter); - const storedRows = getStoredRoomRowsFromDatabase(db); - - for (const stored of storedRows) { - const group = buildRegisteredGroupFromStoredSettings( - db, - stored, - requestedAgentType, - ); - if (group) { - const { jid, ...rest } = group; - result[jid] = rest; - } - } - - return result; + return getAllRoomBindingsFromDatabase(db, agentTypeFilter); } export function getRegisteredAgentTypesForJid(jid: string): AgentType[] { if (!db) return []; - const stored = getStoredRoomSettingsRowFromDatabase(db, jid); - return stored ? resolveStoredRoomCapabilityTypes(db, stored) : []; + return getRegisteredAgentTypesForJidFromDatabase(db, jid); } export function getStoredRoomSettings( chatJid: string, ): StoredRoomSettings | undefined { if (!db) return undefined; - return getStoredRoomSettingsRowFromDatabase(db, chatJid); -} - -function getStoredRoomModeRow(chatJid: string): StoredRoomModeRow | undefined { - const row = getStoredRoomSettings(chatJid); - return row - ? { - roomMode: row.roomMode, - source: row.modeSource, - } - : undefined; -} - -function assertNoPendingLegacyRoomMigration(): void { - const pendingLegacyRows = countPendingLegacyRegisteredGroupRows(db); - if (pendingLegacyRows === 0) { - return; - } - - throw new Error( - `Legacy room migration required before startup (pending_rows=${pendingLegacyRows})`, - ); -} - -function assertNoUnexpectedDataStateFiles(): void { - const pendingFiles = listUnexpectedDataStateFiles(DATA_DIR); - if (pendingFiles.length === 0) { - return; - } - - throw new Error( - `Unexpected data state files detected before startup (files=${pendingFiles.join(',')})`, - ); -} - -function assertNoUnsupportedRouterStateDbKeys(): void { - const legacyKeys = getUnsupportedRouterStateKeysFromDatabase(db); - if (legacyKeys.length === 0) { - return; - } - - throw new Error( - `Unsupported router_state DB keys remain before startup (keys=${legacyKeys.join(',')})`, - ); -} - -function buildRoomRegistrationSnapshotFromStoredRoom( - stored: StoredRoomSettings, - overrides?: Partial>, -): RoomRegistrationSnapshot { - const capabilityTypes = resolveStoredRoomCapabilityTypes(db, stored); - const ownerAgentType = - stored.ownerAgentType || - (capabilityTypes.length > 0 - ? inferOwnerAgentTypeFromRegisteredAgentTypes(capabilityTypes) - : OWNER_AGENT_TYPE); - const name = overrides?.name ?? stored.name ?? stored.chatJid; - - return { - name, - folder: resolveAssignedRoomFolder(db, stored.chatJid, name, stored.folder), - triggerPattern: stored.trigger ?? '', - requiresTrigger: stored.requiresTrigger ?? false, - isMain: stored.isMain ?? false, - ownerAgentType, - workDir: stored.workDir ?? null, - }; -} - -function buildRoomRegistrationPlanForJid( - chatJid: string, - overrides?: Partial>, -): - | { - snapshot: RoomRegistrationSnapshot; - roomMode: RoomMode; - hasStoredRoom: boolean; - } - | undefined { - const stored = getStoredRoomSettingsRowFromDatabase(db, chatJid); - if (stored) { - return { - snapshot: buildRoomRegistrationSnapshotFromStoredRoom(stored, overrides), - roomMode: stored.roomMode, - hasStoredRoom: true, - }; - } - - return undefined; -} - -function upsertStoredRoomMode( - chatJid: string, - roomMode: RoomMode, - source: RoomModeSource, -): void { - db.prepare( - `INSERT INTO room_settings ( - chat_jid, - room_mode, - mode_source, - updated_at - ) VALUES (?, ?, ?, ?) - ON CONFLICT(chat_jid) DO UPDATE SET - room_mode = excluded.room_mode, - mode_source = excluded.mode_source, - updated_at = excluded.updated_at`, - ).run(chatJid, roomMode, source, new Date().toISOString()); + return getStoredRoomSettingsFromDatabase(db, chatJid); } export function getExplicitRoomMode(chatJid: string): RoomMode | undefined { - const row = getStoredRoomModeRow(chatJid); - return row?.source === 'explicit' ? row.roomMode : undefined; + return getExplicitRoomModeFromDatabase(db, chatJid); } export function setExplicitRoomMode(chatJid: string, roomMode: RoomMode): void { - upsertStoredRoomMode(chatJid, roomMode, 'explicit'); - deleteLegacyRegisteredGroupRowsForJid(db, chatJid); + setExplicitRoomModeInDatabase(db, chatJid, roomMode); } export function clearExplicitRoomMode(chatJid: string): void { - const stored = getStoredRoomSettingsRowFromDatabase(db, chatJid); - const agentTypes = stored - ? inferStoredRoomCapabilityTypes(db, stored) - : collectRegisteredAgentTypes(db, chatJid); - if (agentTypes.length === 0) { - db.prepare('DELETE FROM room_settings WHERE chat_jid = ?').run(chatJid); - deleteLegacyRegisteredGroupRowsForJid(db, chatJid); - return; - } - upsertStoredRoomMode( - chatJid, - inferRoomModeFromRegisteredAgentTypes(agentTypes), - 'inferred', - ); - deleteLegacyRegisteredGroupRowsForJid(db, chatJid); + clearExplicitRoomModeInDatabase(db, chatJid); } export function getEffectiveRoomMode(chatJid: string): RoomMode { - return getStoredRoomModeRow(chatJid)?.roomMode ?? 'single'; + return getEffectiveRoomModeFromDatabase(db, chatJid); } export function getEffectiveRuntimeRoomMode(chatJid: string): RoomMode { - return getStoredRoomSettings(chatJid)?.roomMode ?? 'single'; + return getEffectiveRuntimeRoomModeFromDatabase(db, chatJid); } // --- Paired task/project/workspace state --- diff --git a/src/db/database-lifecycle.ts b/src/db/database-lifecycle.ts new file mode 100644 index 0000000..87ae522 --- /dev/null +++ b/src/db/database-lifecycle.ts @@ -0,0 +1,83 @@ +import { Database } from 'bun:sqlite'; + +import { DATA_DIR, normalizeServiceId, SERVICE_ID } from '../config.js'; +import { listUnexpectedDataStateFiles } from '../data-state-files.js'; +import { + openDatabaseFromFile, + openInMemoryDatabase, + openPersistentDatabase, + initializeDatabaseSchema, +} from './bootstrap.js'; +import { countPendingLegacyRegisteredGroupRows } from './room-registration.js'; +import { + clearExpiredPairedTaskExecutionLeasesInDatabase, + clearPairedTaskExecutionLeasesForServiceInDatabase, +} from './paired-state.js'; +import { getUnsupportedRouterStateKeysFromDatabase } from './router-state.js'; + +function finalizeDatabaseInitialization(database: Database): void { + clearPairedTaskExecutionLeasesForServiceInDatabase( + database, + normalizeServiceId(SERVICE_ID), + ); + clearExpiredPairedTaskExecutionLeasesInDatabase(database); +} + +function assertNoPendingLegacyRoomMigration(database: Database): void { + const pendingLegacyRows = countPendingLegacyRegisteredGroupRows(database); + if (pendingLegacyRows === 0) { + return; + } + + throw new Error( + `Legacy room migration required before startup (pending_rows=${pendingLegacyRows})`, + ); +} + +function assertNoUnexpectedDataStateFiles(): void { + const pendingFiles = listUnexpectedDataStateFiles(DATA_DIR); + if (pendingFiles.length === 0) { + return; + } + + throw new Error( + `Unexpected data state files detected before startup (files=${pendingFiles.join(',')})`, + ); +} + +function assertNoUnsupportedRouterStateDbKeys(database: Database): void { + const unsupportedKeys = getUnsupportedRouterStateKeysFromDatabase(database); + if (unsupportedKeys.length === 0) { + return; + } + + throw new Error( + `Unsupported router_state DB keys remain before startup (keys=${unsupportedKeys.join(',')})`, + ); +} + +export function openInitializedPersistentDatabase(): Database { + const database = openPersistentDatabase(); + initializeDatabaseSchema(database); + assertNoPendingLegacyRoomMigration(database); + assertNoUnexpectedDataStateFiles(); + assertNoUnsupportedRouterStateDbKeys(database); + finalizeDatabaseInitialization(database); + return database; +} + +export function openInitializedInMemoryDatabase(): Database { + const database = openInMemoryDatabase(); + initializeDatabaseSchema(database); + finalizeDatabaseInitialization(database); + return database; +} + +export function openInitializedDatabaseFromFile(dbPath: string): Database { + const database = openDatabaseFromFile(dbPath); + initializeDatabaseSchema(database); + assertNoPendingLegacyRoomMigration(database); + assertNoUnsupportedRouterStateDbKeys(database); + finalizeDatabaseInitialization(database); + return database; +} diff --git a/src/db/rooms.ts b/src/db/rooms.ts new file mode 100644 index 0000000..57ca5f0 --- /dev/null +++ b/src/db/rooms.ts @@ -0,0 +1,499 @@ +import { Database } from 'bun:sqlite'; + +import { OWNER_AGENT_TYPE } from '../config.js'; +import { isValidGroupFolder } from '../group-folder.js'; +import type { AgentType, RegisteredGroup, RoomMode } from '../types.js'; +import { + type RoomModeSource, + type RoomRegistrationSnapshot, + type StoredRoomSettings, + buildLegacyRoomMigrationPlan, + buildRegisteredGroupFromStoredSettings, + collectRegisteredAgentTypes, + countPendingLegacyRegisteredGroupRows, + deleteLegacyRegisteredGroupRowsForJid, + getPendingLegacyRegisteredGroupJids as getPendingLegacyRegisteredGroupJidsFromDatabase, + getStoredRoomRowsFromDatabase, + getStoredRoomSettingsRowFromDatabase, + inferOwnerAgentTypeFromRegisteredAgentTypes, + inferRoomModeFromRegisteredAgentTypes, + inferStoredRoomCapabilityTypes, + insertStoredRoomSettings, + insertStoredRoomSettingsFromMigration, + materializeRegisteredGroupsForRoom, + normalizeStoredAgentType, + resolveAssignedRoomFolder, + resolveStoredRoomCapabilityTypes, + syncRoomRoleOverridesForRoom, + updateStoredRoomMetadata, + upsertRoomRoleOverride, +} from './room-registration.js'; + +interface StoredRoomModeRow { + roomMode: RoomMode; + source: RoomModeSource; +} + +export interface AssignRoomInput { + name: string; + roomMode?: RoomMode; + ownerAgentType?: AgentType; + folder?: string; + isMain?: boolean; + workDir?: string; + addedAt?: string; + ownerAgentConfig?: RegisteredGroup['agentConfig']; +} + +export function setStoredRoomOwnerAgentTypeForTestsInDatabase( + database: Database, + chatJid: string, + ownerAgentType: AgentType | null, +): void { + database + .prepare( + `UPDATE room_settings + SET owner_agent_type = ?, + updated_at = ? + WHERE chat_jid = ?`, + ) + .run(ownerAgentType, new Date().toISOString(), chatJid); + if (ownerAgentType) { + const now = new Date().toISOString(); + upsertRoomRoleOverride(database, chatJid, { + role: 'owner', + agentType: ownerAgentType, + createdAt: now, + updatedAt: now, + }); + } +} + +export function deleteStoredRoomSettingsForTestsInDatabase( + database: Database, + chatJid: string, +): void { + database.prepare('DELETE FROM room_settings WHERE chat_jid = ?').run(chatJid); +} + +export function getRegisteredGroupFromDatabase( + database: Database, + jid: string, + agentType?: string, +): (RegisteredGroup & { jid: string }) | undefined { + const requestedAgentType = normalizeStoredAgentType(agentType); + const stored = getStoredRoomSettingsRowFromDatabase(database, jid); + if (!stored) { + return undefined; + } + + return buildRegisteredGroupFromStoredSettings( + database, + stored, + requestedAgentType, + ); +} + +function writeLegacyRegisteredGroupAndSyncRoomSettingsInDatabase( + database: Database, + jid: string, + group: RegisteredGroup, +): void { + const existingStored = getStoredRoomSettingsRowFromDatabase(database, jid); + const existingRoomMode = getStoredRoomModeRowFromDatabase(database, jid); + if (!isValidGroupFolder(group.folder)) { + throw new Error(`Invalid group folder "${group.folder}" for JID ${jid}`); + } + + database.transaction(() => { + const seededAgentType = group.agentType || 'claude-code'; + const agentTypes = new Set( + collectRegisteredAgentTypes(database, jid), + ); + agentTypes.add(seededAgentType); + const inferredRoomMode = inferRoomModeFromRegisteredAgentTypes([ + ...agentTypes, + ]); + const roomMode = + existingRoomMode?.source === 'explicit' + ? existingRoomMode.roomMode + : inferredRoomMode; + const ownerAgentType = + existingStored?.modeSource === 'explicit' && existingStored.ownerAgentType + ? existingStored.ownerAgentType + : inferOwnerAgentTypeFromRegisteredAgentTypes([...agentTypes]); + const snapshot: RoomRegistrationSnapshot = { + name: group.name, + folder: group.folder, + triggerPattern: + existingStored?.modeSource === 'explicit' && existingStored.trigger + ? existingStored.trigger + : (group.trigger ?? ''), + requiresTrigger: + group.requiresTrigger ?? existingStored?.requiresTrigger ?? false, + isMain: group.isMain ?? existingStored?.isMain ?? false, + ownerAgentType, + workDir: group.workDir ?? existingStored?.workDir ?? null, + }; + + if (existingStored) { + updateStoredRoomMetadata(database, jid, snapshot); + if (!existingRoomMode || existingRoomMode.source === 'inferred') { + upsertStoredRoomModeInDatabase(database, jid, roomMode, 'inferred'); + } + } else { + insertStoredRoomSettings(database, jid, roomMode, 'inferred', snapshot); + } + + materializeRegisteredGroupsForRoom( + database, + jid, + snapshot, + roomMode, + ownerAgentType, + seededAgentType === ownerAgentType ? group.agentConfig : undefined, + group.added_at, + ); + syncRoomRoleOverridesForRoom( + database, + jid, + roomMode, + ownerAgentType, + seededAgentType === ownerAgentType ? group.agentConfig : undefined, + group.added_at, + ); + })(); +} + +export function setRegisteredGroupForTestsInDatabase( + database: Database, + jid: string, + group: RegisteredGroup, +): void { + writeLegacyRegisteredGroupAndSyncRoomSettingsInDatabase(database, jid, group); +} + +export function migrateLegacyRoomRegistrationsForTestsInDatabase( + database: Database, +): { + migratedRooms: number; + migratedRoleOverrides: number; +} { + const jids = getPendingLegacyRegisteredGroupJidsFromDatabase(database); + let migratedRooms = 0; + let migratedRoleOverrides = 0; + + database.transaction(() => { + for (const jid of jids) { + const plan = buildLegacyRoomMigrationPlan(database, jid); + if (!plan) continue; + insertStoredRoomSettingsFromMigration(database, plan); + for (const override of plan.roleOverrides) { + upsertRoomRoleOverride(database, jid, override); + migratedRoleOverrides += 1; + } + migratedRooms += 1; + } + })(); + + return { migratedRooms, migratedRoleOverrides }; +} + +export function assignRoomInDatabase( + database: Database, + chatJid: string, + input: AssignRoomInput, +): (RegisteredGroup & { jid: string }) | undefined { + const existing = getStoredRoomSettingsRowFromDatabase(database, chatJid); + const roomMode = input.roomMode || existing?.roomMode || 'single'; + const ownerAgentType = + input.ownerAgentType || existing?.ownerAgentType || OWNER_AGENT_TYPE; + const folder = resolveAssignedRoomFolder( + database, + chatJid, + input.name, + input.folder, + ); + const snapshot: RoomRegistrationSnapshot = { + name: input.name, + folder, + triggerPattern: existing?.trigger ?? '', + requiresTrigger: existing?.requiresTrigger ?? false, + isMain: input.isMain ?? existing?.isMain ?? false, + ownerAgentType, + workDir: input.workDir ?? existing?.workDir ?? null, + }; + const now = new Date().toISOString(); + + database.transaction(() => { + if (existing) { + database + .prepare( + `UPDATE room_settings + SET room_mode = ?, + mode_source = 'explicit', + name = ?, + folder = ?, + trigger_pattern = ?, + requires_trigger = ?, + is_main = ?, + owner_agent_type = ?, + work_dir = ?, + updated_at = ? + WHERE chat_jid = ?`, + ) + .run( + roomMode, + snapshot.name, + snapshot.folder, + snapshot.triggerPattern, + snapshot.requiresTrigger ? 1 : 0, + snapshot.isMain ? 1 : 0, + snapshot.ownerAgentType, + snapshot.workDir, + now, + chatJid, + ); + } else { + insertStoredRoomSettings( + database, + chatJid, + roomMode, + 'explicit', + snapshot, + ); + } + + syncRoomRoleOverridesForRoom( + database, + chatJid, + roomMode, + ownerAgentType, + input.ownerAgentConfig, + input.addedAt ?? now, + ); + deleteLegacyRegisteredGroupRowsForJid(database, chatJid); + })(); + + return getRegisteredGroupFromDatabase(database, chatJid); +} + +export function updateRegisteredGroupNameInDatabase( + database: Database, + jid: string, + name: string, +): void { + const plan = buildRoomRegistrationPlanForJid(database, jid, { name }); + if (!plan) { + return; + } + + database.transaction(() => { + if (plan.hasStoredRoom) { + updateStoredRoomMetadata(database, jid, plan.snapshot); + } else { + insertStoredRoomSettings( + database, + jid, + plan.roomMode, + 'inferred', + plan.snapshot, + ); + } + deleteLegacyRegisteredGroupRowsForJid(database, jid); + })(); +} + +export function getAllRoomBindingsFromDatabase( + database: Database, + agentTypeFilter?: string, +): Record { + const result: Record = {}; + const requestedAgentType = normalizeStoredAgentType(agentTypeFilter); + const storedRows = getStoredRoomRowsFromDatabase(database); + + for (const stored of storedRows) { + const group = buildRegisteredGroupFromStoredSettings( + database, + stored, + requestedAgentType, + ); + if (!group) continue; + const { jid, ...rest } = group; + result[jid] = rest; + } + + return result; +} + +export function getRegisteredAgentTypesForJidFromDatabase( + database: Database, + jid: string, +): AgentType[] { + const stored = getStoredRoomSettingsRowFromDatabase(database, jid); + return stored ? resolveStoredRoomCapabilityTypes(database, stored) : []; +} + +export function getStoredRoomSettingsFromDatabase( + database: Database, + chatJid: string, +): StoredRoomSettings | undefined { + return getStoredRoomSettingsRowFromDatabase(database, chatJid); +} + +function getStoredRoomModeRowFromDatabase( + database: Database, + chatJid: string, +): StoredRoomModeRow | undefined { + const row = getStoredRoomSettingsFromDatabase(database, chatJid); + return row + ? { + roomMode: row.roomMode, + source: row.modeSource, + } + : undefined; +} + +function buildRoomRegistrationSnapshotFromStoredRoom( + database: Database, + stored: StoredRoomSettings, + overrides?: Partial>, +): RoomRegistrationSnapshot { + const capabilityTypes = resolveStoredRoomCapabilityTypes(database, stored); + const ownerAgentType = + stored.ownerAgentType || + (capabilityTypes.length > 0 + ? inferOwnerAgentTypeFromRegisteredAgentTypes(capabilityTypes) + : OWNER_AGENT_TYPE); + const name = overrides?.name ?? stored.name ?? stored.chatJid; + + return { + name, + folder: resolveAssignedRoomFolder( + database, + stored.chatJid, + name, + stored.folder, + ), + triggerPattern: stored.trigger ?? '', + requiresTrigger: stored.requiresTrigger ?? false, + isMain: stored.isMain ?? false, + ownerAgentType, + workDir: stored.workDir ?? null, + }; +} + +function buildRoomRegistrationPlanForJid( + database: Database, + chatJid: string, + overrides?: Partial>, +): + | { + snapshot: RoomRegistrationSnapshot; + roomMode: RoomMode; + hasStoredRoom: boolean; + } + | undefined { + const stored = getStoredRoomSettingsRowFromDatabase(database, chatJid); + if (!stored) { + return undefined; + } + + return { + snapshot: buildRoomRegistrationSnapshotFromStoredRoom( + database, + stored, + overrides, + ), + roomMode: stored.roomMode, + hasStoredRoom: true, + }; +} + +function upsertStoredRoomModeInDatabase( + database: Database, + chatJid: string, + roomMode: RoomMode, + source: RoomModeSource, +): void { + database + .prepare( + `INSERT INTO room_settings ( + chat_jid, + room_mode, + mode_source, + updated_at + ) VALUES (?, ?, ?, ?) + ON CONFLICT(chat_jid) DO UPDATE SET + room_mode = excluded.room_mode, + mode_source = excluded.mode_source, + updated_at = excluded.updated_at`, + ) + .run(chatJid, roomMode, source, new Date().toISOString()); +} + +export function getExplicitRoomModeFromDatabase( + database: Database, + chatJid: string, +): RoomMode | undefined { + const row = getStoredRoomModeRowFromDatabase(database, chatJid); + return row?.source === 'explicit' ? row.roomMode : undefined; +} + +export function setExplicitRoomModeInDatabase( + database: Database, + chatJid: string, + roomMode: RoomMode, +): void { + upsertStoredRoomModeInDatabase(database, chatJid, roomMode, 'explicit'); + deleteLegacyRegisteredGroupRowsForJid(database, chatJid); +} + +export function clearExplicitRoomModeInDatabase( + database: Database, + chatJid: string, +): void { + const stored = getStoredRoomSettingsRowFromDatabase(database, chatJid); + const agentTypes = stored + ? inferStoredRoomCapabilityTypes(database, stored) + : collectRegisteredAgentTypes(database, chatJid); + if (agentTypes.length === 0) { + database + .prepare('DELETE FROM room_settings WHERE chat_jid = ?') + .run(chatJid); + deleteLegacyRegisteredGroupRowsForJid(database, chatJid); + return; + } + + upsertStoredRoomModeInDatabase( + database, + chatJid, + inferRoomModeFromRegisteredAgentTypes(agentTypes), + 'inferred', + ); + deleteLegacyRegisteredGroupRowsForJid(database, chatJid); +} + +export function getEffectiveRoomModeFromDatabase( + database: Database, + chatJid: string, +): RoomMode { + return ( + getStoredRoomModeRowFromDatabase(database, chatJid)?.roomMode ?? 'single' + ); +} + +export function getEffectiveRuntimeRoomModeFromDatabase( + database: Database, + chatJid: string, +): RoomMode { + return ( + getStoredRoomSettingsFromDatabase(database, chatJid)?.roomMode ?? 'single' + ); +} + +export function countPendingLegacyRegisteredGroupRowsInDatabase( + database: Database, +): number { + return countPendingLegacyRegisteredGroupRows(database); +} diff --git a/src/index.ts b/src/index.ts index cfe736e..41878f1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,3 @@ -import fs from 'fs'; -import path from 'path'; - import { ASSISTANT_NAME, DATA_DIR, @@ -20,27 +17,17 @@ import { getRegisteredChannelNames, } from './channels/registry.js'; import { writeGroupsSnapshot } from './agent-runner.js'; -import { listAvailableGroups } from './available-groups.js'; import { type AssignRoomInput, - assignRoom, - getAllRoomBindings, - getAllSessions, getAllTasks, - getLatestMessageSeqAtOrBefore, hasRecentRestartAnnouncement, - getRouterState, initDatabase, - setRouterState, - deleteAllSessionsForGroup, - deleteSession, - setSession, storeChatMetadata, storeMessage, } from './db.js'; import { composeDashboardContent } from './dashboard-render.js'; import { GroupQueue } from './group-queue.js'; -import { resolveGroupFolderPath, resolveGroupIpcPath } from './group-folder.js'; +import { resolveGroupIpcPath } from './group-folder.js'; import { startIpcWatcher } from './ipc.js'; import { findChannel, @@ -69,7 +56,6 @@ import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.js'; import { startUnifiedDashboard } from './unified-dashboard.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; -import { normalizeStoredSeqCursor } from './message-cursor.js'; import { initCodexTokenRotation } from './codex-token-rotation.js'; import { hasAvailableClaudeToken, @@ -88,6 +74,7 @@ import { clearGlobalFailover, getGlobalFailoverInfo, } from './service-routing.js'; +import { createRuntimeState } from './runtime-state.js'; import { FAILOVER_MIN_DURATION_MS } from './config.js'; // Token rotation is initialized lazily on first use or at startup below @@ -137,13 +124,9 @@ export async function editFormattedTrackedChannelMessage( await channel.editMessage(jid, messageId, text); } -let lastTimestamp = ''; -let sessions: Record = {}; -let roomBindings: Record = {}; -let lastAgentTimestamp: Record = {}; - const channels: Channel[] = []; const queue = new GroupQueue(); +const runtimeState = createRuntimeState(); const runtime = createMessageRuntime({ assistantName: ASSISTANT_NAME, idleTimeout: IDLE_TIMEOUT, @@ -152,109 +135,36 @@ const runtime = createMessageRuntime({ triggerPattern: TRIGGER_PATTERN, channels, queue, - getRoomBindings: () => roomBindings, - getSessions: () => sessions, - getLastTimestamp: () => lastTimestamp, - setLastTimestamp: (timestamp) => { - lastTimestamp = timestamp; - }, - getLastAgentTimestamps: () => lastAgentTimestamp, - saveState, - persistSession: (groupFolder, sessionId) => { - sessions[groupFolder] = sessionId; - setSession(groupFolder, sessionId); - }, - clearSession, + getRoomBindings: runtimeState.getRoomBindings, + getSessions: runtimeState.getSessions, + getLastTimestamp: runtimeState.getLastTimestamp, + setLastTimestamp: runtimeState.setLastTimestamp, + getLastAgentTimestamps: runtimeState.getLastAgentTimestamps, + saveState: runtimeState.saveState, + persistSession: runtimeState.persistSession, + clearSession: runtimeState.clearSession, }); -function loadState(): void { - lastTimestamp = normalizeStoredSeqCursor(getRouterState('last_seq')); - const agentTs = getRouterState('last_agent_seq'); - try { - const parsed = agentTs - ? (JSON.parse(agentTs) as Record) - : {}; - lastAgentTimestamp = Object.fromEntries( - Object.entries(parsed).map(([chatJid, cursor]) => [ - chatJid, - normalizeStoredSeqCursor(cursor, chatJid), - ]), - ); - } catch { - logger.warn('Corrupted last_agent_seq in DB, resetting'); - lastAgentTimestamp = {}; - } - sessions = getAllSessions(); - roomBindings = getAllRoomBindings(); - logger.info( - { - groupCount: Object.keys(roomBindings).length, - agentType: 'unified', - }, - 'State loaded', - ); -} - -function saveState(): void { - setRouterState('last_seq', lastTimestamp); - setRouterState('last_agent_seq', JSON.stringify(lastAgentTimestamp)); -} - -function clearSession( - groupFolder: string, - opts?: { allRoles?: boolean }, -): void { - delete sessions[groupFolder]; - if (opts?.allRoles) { - deleteAllSessionsForGroup(groupFolder); - } else { - deleteSession(groupFolder); - } -} - -function assignRoomForIpc(jid: string, input: AssignRoomInput): void { - const assignedGroup = assignRoom(jid, input); - if (!assignedGroup) { - logger.warn({ jid }, 'Failed to assign room from IPC'); - return; - } - - let groupDir: string; - try { - groupDir = resolveGroupFolderPath(assignedGroup.folder); - } catch (err) { - logger.warn( - { jid, folder: assignedGroup.folder, err }, - 'Rejecting room assignment with invalid folder', - ); - return; - } - - const { jid: _ignoredJid, ...storedGroup } = assignedGroup; - roomBindings[jid] = storedGroup; - fs.mkdirSync(path.join(groupDir, 'logs'), { recursive: true }); -} - /** * Get available groups list for the agent. * Returns groups ordered by most recent activity. */ export function getAvailableGroups(): import('./agent-runner.js').AvailableGroup[] { - return listAvailableGroups(roomBindings); + return runtimeState.getAvailableGroups(); } /** @internal - exported for testing */ export function _setRegisteredGroups( groups: Record, ): void { - roomBindings = groups; + runtimeState.setRoomBindings(groups); } /** @internal - exported for testing */ export function _setRoomBindings( groups: Record, ): void { - roomBindings = groups; + runtimeState.setRoomBindings(groups); } async function announceRestartRecovery( @@ -295,7 +205,10 @@ async function announceRestartRecovery( return explicitContext; } - const inferred = inferRecentRestartContext(roomBindings, processStartedAtMs); + const inferred = inferRecentRestartContext( + runtimeState.getRoomBindings(), + processStartedAtMs, + ); if (!inferred) return null; if (hasRecentRestartAnnouncement(inferred.chatJid, dedupeSince)) { @@ -326,7 +239,7 @@ async function main(): Promise { initCodexTokenRotation(); startTokenRefreshLoop(); - loadState(); + runtimeState.loadState(); // Graceful shutdown handlers let leaseRecoveryTimer: ReturnType | null = null; @@ -337,6 +250,7 @@ async function main(): Promise { clearInterval(leaseRecoveryTimer); leaseRecoveryTimer = null; } + const roomBindings = runtimeState.getRoomBindings(); const interruptedGroups = queue .getStatuses(Object.keys(roomBindings)) .filter( @@ -380,6 +294,7 @@ async function main(): Promise { const channelOpts = { onMessage: (chatJid: string, msg: NewMessage) => { // Sender allowlist drop mode: discard messages from denied senders before storing + const roomBindings = runtimeState.getRoomBindings(); if (!msg.is_from_me && !msg.is_bot_message && roomBindings[chatJid]) { const cfg = loadSenderAllowlist(); if ( @@ -404,7 +319,7 @@ async function main(): Promise { channel?: string, isGroup?: boolean, ) => storeChatMetadata(chatJid, timestamp, name, channel, isGroup), - roomBindings: () => roomBindings, + roomBindings: runtimeState.getRoomBindings, }; // Create and connect all registered channels. @@ -437,8 +352,8 @@ async function main(): Promise { ); startSchedulerLoop({ - roomBindings: () => roomBindings, - getSessions: () => sessions, + roomBindings: runtimeState.getRoomBindings, + getSessions: runtimeState.getSessions, queue, onProcess: (groupJid, proc, processName, ipcDir) => queue.registerProcess(groupJid, proc, processName, ipcDir), @@ -480,8 +395,8 @@ async function main(): Promise { } }, nudgeScheduler: nudgeSchedulerLoop, - roomBindings: () => roomBindings, - assignRoom: assignRoomForIpc, + roomBindings: runtimeState.getRoomBindings, + assignRoom: runtimeState.assignRoomForIpc, syncGroups: async (force: boolean) => { await Promise.all( channels @@ -496,6 +411,7 @@ async function main(): Promise { queue.enterRecoveryMode(); runtime.recoverPendingMessages(); const restartContext = await announceRestartRecovery(processStartedAtMs); + const roomBindings = runtimeState.getRoomBindings(); for (const candidate of getInterruptedRecoveryCandidates( restartContext, roomBindings, @@ -524,9 +440,9 @@ async function main(): Promise { usageUpdateInterval: USAGE_UPDATE_INTERVAL, channels, queue, - roomBindings: () => roomBindings, + roomBindings: runtimeState.getRoomBindings, onGroupNameSynced: (jid, name) => { - const group = roomBindings[jid]; + const group = runtimeState.getRoomBindings()[jid]; if (group) { group.name = name; } diff --git a/src/message-runtime-turns.ts b/src/message-runtime-turns.ts new file mode 100644 index 0000000..5553a19 --- /dev/null +++ b/src/message-runtime-turns.ts @@ -0,0 +1,267 @@ +import { type AgentOutput } from './agent-runner.js'; +import { getLastBotFinalMessage } from './db.js'; +import { runAgentForGroup } from './message-agent-executor.js'; +import { MessageTurnController } from './message-turn-controller.js'; +import { + getEffectiveChannelLease, + hasReviewerLease, + resolveLeaseServiceId, +} from './service-routing.js'; +import { normalizeMessageForDedupe } from './router.js'; +import type { ExecuteTurnFn } from './message-runtime-types.js'; +import type { + Channel, + NewMessage, + AgentType, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; +import type { GroupQueue } from './group-queue.js'; +import type { PairedTurnIdentity } from './paired-turn-identity.js'; +import { logger } from './logger.js'; + +export function isDuplicateOfLastBotFinal( + chatJid: string, + text: string, +): boolean { + if (!hasReviewerLease(chatJid)) { + return false; + } + + const lastMessages = getLastBotFinalMessage(chatJid, 'claude-code', 1); + if (lastMessages.length === 0) { + return false; + } + + const lastMessage = lastMessages[0]; + const normalizedLast = normalizeMessageForDedupe(lastMessage.content); + const normalizedCurrent = normalizeMessageForDedupe(text); + + return normalizedLast === normalizedCurrent && normalizedLast.length > 0; +} + +export function labelPairedSenders( + channels: Channel[], + chatJid: string, + messages: NewMessage[], +): NewMessage[] { + if (!hasReviewerLease(chatJid)) return messages; + + const botIdToChannelName = new Map(); + for (const ch of channels) { + if (!ch.isConnected()) continue; + for (const msg of messages) { + if (msg.is_bot_message && ch.isOwnMessage?.(msg)) { + botIdToChannelName.set(msg.sender, ch.name); + } + } + } + + const channelToRole: Record = { + discord: 'owner', + 'discord-review': 'reviewer', + 'discord-arbiter': 'arbiter', + }; + + return messages.map((msg) => { + if (!msg.is_bot_message) return msg; + const channelName = botIdToChannelName.get(msg.sender); + if (!channelName) return msg; + const role = channelToRole[channelName]; + return role ? { ...msg, sender_name: role } : msg; + }); +} + +interface CreateExecuteTurnDeps { + runAgent: ( + group: RegisteredGroup, + prompt: string, + chatJid: string, + runId: string, + onOutput?: (output: AgentOutput) => Promise, + options?: { + startSeq?: number | null; + endSeq?: number | null; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; + pairedTurnIdentity?: PairedTurnIdentity; + }, + ) => Promise<'success' | 'error'>; + assistantName: string; + idleTimeout: number; + failureFinalText: string; + channels: Channel[]; + queue: GroupQueue; + getRoomBindings: () => Record; + getSessions: () => Record; + persistSession: (groupFolder: string, sessionId: string) => void; + clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; + deliverFinalText: (args: { + text: string; + chatJid: string; + runId: string; + channel: Channel; + group: RegisteredGroup; + startSeq: number | null; + endSeq: number | null; + forcedAgentType?: AgentType; + deliveryRole: PairedRoomRole | null; + deliveryServiceId: string | null; + }) => Promise; + afterDeliverySuccess?: (args: { + chatJid: string; + runId: string; + deliveryRole: PairedRoomRole | null; + pairedRoom: boolean; + }) => Promise; +} + +export function createRunAgent(deps: { + assistantName: string; + queue: GroupQueue; + getRoomBindings: () => Record; + getSessions: () => Record; + persistSession: (groupFolder: string, sessionId: string) => void; + clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; +}) { + return async ( + group: RegisteredGroup, + prompt: string, + chatJid: string, + runId: string, + onOutput?: (output: AgentOutput) => Promise, + options?: { + startSeq?: number | null; + endSeq?: number | null; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; + pairedTurnIdentity?: PairedTurnIdentity; + }, + ): Promise<'success' | 'error'> => + runAgentForGroup(deps, { + group, + prompt, + chatJid, + runId, + startSeq: options?.startSeq, + endSeq: options?.endSeq, + hasHumanMessage: options?.hasHumanMessage, + forcedRole: options?.forcedRole, + forcedAgentType: options?.forcedAgentType, + pairedTurnIdentity: options?.pairedTurnIdentity, + onOutput, + }); +} + +export function createExecuteTurn(deps: CreateExecuteTurnDeps): ExecuteTurnFn { + return async (args) => { + const { group, prompt, chatJid, runId, channel, startSeq, endSeq } = args; + const isClaudeCodeAgent = + (args.forcedAgentType ?? group.agentType ?? 'claude-code') === + 'claude-code'; + const pairedRoom = hasReviewerLease(chatJid); + const resolvedDeliveryRole = + args.deliveryRole ?? args.forcedRole ?? (pairedRoom ? 'owner' : null); + const resolvedDeliveryServiceId = resolveLeaseServiceId( + getEffectiveChannelLease(chatJid), + resolvedDeliveryRole ?? 'owner', + ); + const allowProgressReplayWithoutFinal = + args.pairedTurnIdentity?.role !== 'reviewer' && + args.pairedTurnIdentity?.role !== 'arbiter'; + const turnController = new MessageTurnController({ + chatJid, + group, + runId, + channel, + idleTimeout: deps.idleTimeout, + failureFinalText: deps.failureFinalText, + isClaudeCodeAgent, + clearSession: () => deps.clearSession(group.folder), + requestClose: (reason) => + deps.queue.closeStdin(chatJid, { runId, reason }), + allowProgressReplayWithoutFinal, + deliveryRole: resolvedDeliveryRole, + deliveryServiceId: resolvedDeliveryServiceId, + pairedTurnIdentity: args.pairedTurnIdentity ?? null, + deliverFinalText: async (text) => { + try { + return await deps.deliverFinalText({ + text, + chatJid, + runId, + channel, + group, + startSeq, + endSeq, + forcedAgentType: args.forcedAgentType, + deliveryRole: resolvedDeliveryRole, + deliveryServiceId: resolvedDeliveryServiceId, + }); + } catch (err) { + logger.warn( + { group: group.name, chatJid, runId, err }, + 'Failed to persist produced output for delivery', + ); + return false; + } + }, + }); + + await turnController.start(); + + try { + const outputStatus = await deps.runAgent( + group, + prompt, + chatJid, + runId, + async (result) => { + await turnController.handleOutput(result); + }, + { + startSeq, + endSeq, + hasHumanMessage: args.hasHumanMessage, + forcedRole: args.forcedRole, + forcedAgentType: args.forcedAgentType, + pairedTurnIdentity: args.pairedTurnIdentity, + }, + ); + + const { deliverySucceeded, visiblePhase } = + await turnController.finish(outputStatus); + + if (deliverySucceeded) { + await deps.afterDeliverySuccess?.({ + chatJid, + runId, + deliveryRole: resolvedDeliveryRole, + pairedRoom, + }); + } + + return { + outputStatus, + deliverySucceeded, + visiblePhase, + }; + } finally { + turnController.cancelPendingTypingDelay(); + logger.debug( + { + transition: 'typing:off', + source: 'message-runtime:safety-net', + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + }, + 'Typing indicator transition', + ); + await channel.setTyping?.(chatJid, false); + } + }; +} diff --git a/src/message-runtime.ts b/src/message-runtime.ts index d17ee62..0a7cd41 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,10 +1,7 @@ -import { AgentOutput } from './agent-runner.js'; -import { getAgentOutputText } from './agent-output.js'; import { createProducedWorkItem, getOpenWorkItemForChat, getMessagesSinceSeq, - getLastBotFinalMessage, getLatestOpenPairedTaskForChat, getPairedTaskById, } from './db.js'; @@ -13,12 +10,7 @@ import { SERVICE_SESSION_SCOPE, } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; -import { - findChannel, - findChannelByName, - formatMessages, - normalizeMessageForDedupe, -} from './router.js'; +import { findChannel, findChannelByName, formatMessages } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js'; import { @@ -50,25 +42,17 @@ import { enqueuePairedFollowUpAfterEvent, schedulePairedFollowUpWithMessageCheck, } from './message-runtime-follow-up.js'; -import { runAgentForGroup } from './message-agent-executor.js'; -import { MessageTurnController } from './message-turn-controller.js'; -import type { PairedTurnIdentity } from './paired-turn-identity.js'; import { type ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; import { transitionPairedTaskStatus } from './paired-execution-context-shared.js'; import { Channel, NewMessage, - type AgentType, type PairedRoomRole, RegisteredGroup, type PairedTask, } from './types.js'; import { createScopedLogger, logger } from './logger.js'; -import { - getEffectiveChannelLease, - hasReviewerLease, - resolveLeaseServiceId, -} from './service-routing.js'; +import { hasReviewerLease } from './service-routing.js'; export { resolveHandoffCursorKey, resolveHandoffRoleOverride, @@ -77,31 +61,14 @@ import { getFixedRoleChannelName, getMissingRoleChannelMessage, } from './message-runtime-shared.js'; +import { + createRunAgent, + createExecuteTurn, + isDuplicateOfLastBotFinal, + labelPairedSenders, +} from './message-runtime-turns.js'; -/** - * Check if a message is a duplicate of the last bot final message in a paired room. - * Exported for testing purposes. - */ -export function isDuplicateOfLastBotFinal( - chatJid: string, - text: string, -): boolean { - if (!hasReviewerLease(chatJid)) { - return false; - } - - // Get the last bot final message from DB (any bot, not just this service) - const lastMessages = getLastBotFinalMessage(chatJid, 'claude-code', 1); - if (lastMessages.length === 0) { - return false; - } - - const lastMessage = lastMessages[0]; - const normalizedLast = normalizeMessageForDedupe(lastMessage.content); - const normalizedCurrent = normalizeMessageForDedupe(text); - - return normalizedLast === normalizedCurrent && normalizedLast.length > 0; -} +export { isDuplicateOfLastBotFinal }; export interface MessageRuntimeDeps { assistantName: string; @@ -134,37 +101,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const continuationTracker = createImplicitContinuationTracker( deps.idleTimeout, ); - // In paired rooms, replace bot sender_name with role label so agents - // know who is the owner and who is the reviewer regardless of bot nickname. - const labelPairedSenders = ( + const labelPairedRuntimeSenders = ( chatJid: string, messages: NewMessage[], - ): NewMessage[] => { - if (!hasReviewerLease(chatJid)) return messages; - // Build bot-user-id → channel-name mapping from connected channels - const botIdToChannelName = new Map(); - for (const ch of deps.channels) { - if (!ch.isConnected()) continue; - // Probe each bot message to find which channel owns it - for (const msg of messages) { - if (msg.is_bot_message && ch.isOwnMessage?.(msg)) { - botIdToChannelName.set(msg.sender, ch.name); - } - } - } - const channelToRole: Record = { - discord: 'owner', - 'discord-review': 'reviewer', - 'discord-arbiter': 'arbiter', - }; - return messages.map((msg) => { - if (!msg.is_bot_message) return msg; - const channelName = botIdToChannelName.get(msg.sender); - if (!channelName) return msg; - const role = channelToRole[channelName]; - return role ? { ...msg, sender_name: role } : msg; - }); - }; + ): NewMessage[] => labelPairedSenders(deps.channels, chatJid, messages); const enqueueScopedGroupMessageCheck = buildScopedMessageCheckEnqueuer( deps.queue, @@ -197,230 +137,115 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ): boolean => { return isDuplicateOfLastBotFinal(chatJid, text); }; - - const runAgent = async ( - group: RegisteredGroup, - prompt: string, - chatJid: string, - runId: string, - onOutput?: (output: AgentOutput) => Promise, - options?: { - startSeq?: number | null; - endSeq?: number | null; - hasHumanMessage?: boolean; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; - pairedTurnIdentity?: PairedTurnIdentity; - }, - ): Promise<'success' | 'error'> => - runAgentForGroup( - { - assistantName: deps.assistantName, - queue: deps.queue, - getRoomBindings: deps.getRoomBindings, - getSessions: deps.getSessions, - persistSession: deps.persistSession, - clearSession: deps.clearSession, - }, - { - group, - prompt, - chatJid, - runId, - startSeq: options?.startSeq, - endSeq: options?.endSeq, - hasHumanMessage: options?.hasHumanMessage, - forcedRole: options?.forcedRole, - forcedAgentType: options?.forcedAgentType, - pairedTurnIdentity: options?.pairedTurnIdentity, - onOutput, - }, - ); - - const executeTurn = async (args: { - group: RegisteredGroup; - prompt: string; - chatJid: string; - runId: string; - channel: Channel; - startSeq: number | null; - endSeq: number | null; - deliveryRole?: PairedRoomRole; - hasHumanMessage?: boolean; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; - pairedTurnIdentity?: PairedTurnIdentity; - }): Promise<{ - outputStatus: 'success' | 'error'; - deliverySucceeded: boolean; - visiblePhase: ReturnType extends Promise< - infer T - > - ? T extends { visiblePhase: infer V } - ? V - : never - : never; - }> => { - const { group, prompt, chatJid, runId, channel, startSeq, endSeq } = args; - const isClaudeCodeAgent = - (args.forcedAgentType ?? group.agentType ?? 'claude-code') === - 'claude-code'; - const pairedRoom = hasReviewerLease(chatJid); - const resolvedDeliveryRole = - args.deliveryRole ?? args.forcedRole ?? (pairedRoom ? 'owner' : null); - const resolvedDeliveryServiceId = resolveLeaseServiceId( - getEffectiveChannelLease(chatJid), - resolvedDeliveryRole ?? 'owner', - ); - const allowProgressReplayWithoutFinal = - args.pairedTurnIdentity?.role !== 'reviewer' && - args.pairedTurnIdentity?.role !== 'arbiter'; - const turnController = new MessageTurnController({ + const runAgent = createRunAgent({ + assistantName: deps.assistantName, + queue: deps.queue, + getRoomBindings: deps.getRoomBindings, + getSessions: deps.getSessions, + persistSession: deps.persistSession, + clearSession: deps.clearSession, + }); + const executeTurn = createExecuteTurn({ + runAgent, + assistantName: deps.assistantName, + idleTimeout: deps.idleTimeout, + failureFinalText: FAILURE_FINAL_TEXT, + channels: deps.channels, + queue: deps.queue, + getRoomBindings: deps.getRoomBindings, + getSessions: deps.getSessions, + persistSession: deps.persistSession, + clearSession: deps.clearSession, + deliverFinalText: async ({ + text, chatJid, - group, runId, channel, - idleTimeout: deps.idleTimeout, - failureFinalText: FAILURE_FINAL_TEXT, - isClaudeCodeAgent, - clearSession: () => deps.clearSession(group.folder), - requestClose: (reason) => - deps.queue.closeStdin(chatJid, { runId, reason }), - allowProgressReplayWithoutFinal, - deliveryRole: resolvedDeliveryRole, - deliveryServiceId: resolvedDeliveryServiceId, - pairedTurnIdentity: args.pairedTurnIdentity ?? null, - deliverFinalText: async (text) => { - try { - const persistedDeliveryRole = resolvedDeliveryRole; - const persistedDeliveryServiceId = resolvedDeliveryServiceId; - if ( - (persistedDeliveryRole === 'reviewer' || - persistedDeliveryRole === 'arbiter') && - deps.queue.hasDirectTerminalDeliveryForRun?.( - chatJid, - runId, - persistedDeliveryRole, - ) - ) { - logger.info( - { - chatJid, - runId, - deliveryRole: persistedDeliveryRole, - }, - 'Skipping final work item delivery because this run already sent a direct terminal IPC message', - ); - return true; - } - const workItem = createProducedWorkItem({ - group_folder: group.folder, - chat_jid: chatJid, - agent_type: - args.forcedAgentType ?? group.agentType ?? 'claude-code', - service_id: persistedDeliveryServiceId ?? undefined, - delivery_role: persistedDeliveryRole, - start_seq: startSeq, - end_seq: endSeq, - result_payload: text, - }); - return deliverOpenWorkItem({ - channel, - item: workItem, - log: logger, - isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, - openContinuation: (targetChatJid) => - continuationTracker.open(targetChatJid), - }); - } catch (err) { - logger.warn( - { group: group.name, chatJid, runId, err }, - 'Failed to persist produced output for delivery', - ); - return false; - } - }, - }); - - await turnController.start(); - - try { - const outputStatus = await runAgent( - group, - prompt, + group, + startSeq, + endSeq, + forcedAgentType, + deliveryRole, + deliveryServiceId, + }) => { + if ( + (deliveryRole === 'reviewer' || deliveryRole === 'arbiter') && + deps.queue.hasDirectTerminalDeliveryForRun?.( + chatJid, + runId, + deliveryRole, + ) + ) { + logger.info( + { + chatJid, + runId, + deliveryRole, + }, + 'Skipping final work item delivery because this run already sent a direct terminal IPC message', + ); + return true; + } + const workItem = createProducedWorkItem({ + group_folder: group.folder, + chat_jid: chatJid, + agent_type: forcedAgentType ?? group.agentType ?? 'claude-code', + service_id: deliveryServiceId ?? undefined, + delivery_role: deliveryRole, + start_seq: startSeq, + end_seq: endSeq, + result_payload: text, + }); + return deliverOpenWorkItem({ + channel, + item: workItem, + log: logger, + isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, + openContinuation: (targetChatJid) => + continuationTracker.open(targetChatJid), + }); + }, + afterDeliverySuccess: async ({ + chatJid, + runId, + deliveryRole, + pairedRoom, + }) => { + if (!deliveryRole || !pairedRoom) { + return; + } + const pendingTaskAfterDelivery = getLatestOpenPairedTaskForChat(chatJid); + const followUpResult = enqueuePairedFollowUpAfterEvent({ chatJid, runId, - async (result) => { - await turnController.handleOutput(result); - }, - { - startSeq, - endSeq, - hasHumanMessage: args.hasHumanMessage, - forcedRole: args.forcedRole, - forcedAgentType: args.forcedAgentType, - pairedTurnIdentity: args.pairedTurnIdentity, - }, - ); - - const { deliverySucceeded, visiblePhase } = - await turnController.finish(outputStatus); - - if (deliverySucceeded && pairedRoom && resolvedDeliveryRole) { - const pendingTaskAfterDelivery = - getLatestOpenPairedTaskForChat(chatJid); - const followUpResult = enqueuePairedFollowUpAfterEvent({ - chatJid, - runId, - task: pendingTaskAfterDelivery, - source: 'delivery-success', - completedRole: resolvedDeliveryRole, - fallbackLastTurnOutputRole: resolvedDeliveryRole, - enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), - }); - if (followUpResult.kind === 'paired-follow-up') { - logger.info( - { - chatJid, - runId, - completedRole: resolvedDeliveryRole, - taskId: followUpResult.taskId, - taskStatus: followUpResult.taskStatus, - intentKind: followUpResult.intentKind, - scheduled: followUpResult.scheduled, - }, - followUpResult.scheduled - ? resolvedDeliveryRole === 'owner' - ? 'Queued paired follow-up after successful owner delivery' - : 'Queued paired follow-up after successful reviewer/arbiter delivery' - : resolvedDeliveryRole === 'owner' - ? 'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged' - : 'Skipped duplicate paired follow-up after successful reviewer/arbiter delivery while task state was unchanged', - ); - } + task: pendingTaskAfterDelivery, + source: 'delivery-success', + completedRole: deliveryRole, + fallbackLastTurnOutputRole: deliveryRole, + enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), + }); + if (followUpResult.kind === 'paired-follow-up') { + logger.info( + { + chatJid, + runId, + completedRole: deliveryRole, + taskId: followUpResult.taskId, + taskStatus: followUpResult.taskStatus, + intentKind: followUpResult.intentKind, + scheduled: followUpResult.scheduled, + }, + followUpResult.scheduled + ? deliveryRole === 'owner' + ? 'Queued paired follow-up after successful owner delivery' + : 'Queued paired follow-up after successful reviewer/arbiter delivery' + : deliveryRole === 'owner' + ? 'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged' + : 'Skipped duplicate paired follow-up after successful reviewer/arbiter delivery while task state was unchanged', + ); } - - return { - outputStatus, - deliverySucceeded, - visiblePhase, - }; - } finally { - turnController.cancelPendingTypingDelay(); - logger.debug( - { - transition: 'typing:off', - source: 'message-runtime:safety-net', - chatJid, - group: group.name, - groupFolder: group.folder, - runId, - }, - 'Typing indicator transition', - ); - await channel.setTyping?.(chatJid, false); - } - }; + }, + }); const enqueuePendingHandoffs = (): void => { enqueueClaimedServiceHandoffs({ @@ -568,7 +393,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { executeTurn, getFixedRoleChannelName, roleToChannel, - labelPairedSenders, + labelPairedSenders: labelPairedRuntimeSenders, mode: 'idle', }); if (pendingTurnOutcome !== null) { @@ -602,7 +427,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { executeTurn, getFixedRoleChannelName, roleToChannel, - labelPairedSenders, + labelPairedSenders: labelPairedRuntimeSenders, mode: 'bot-only', missedMessages, }); @@ -697,7 +522,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { saveState: deps.saveState, executeTurn, getFixedRoleChannelName, - labelPairedSenders, + labelPairedSenders: labelPairedRuntimeSenders, formatMessages, }); } @@ -735,7 +560,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.queue.closeStdin(chatJid, { reason, }), - labelPairedSenders, + labelPairedSenders: labelPairedRuntimeSenders, }); } catch (err) { logger.error({ err }, 'Error in message loop'); diff --git a/src/runtime-state.ts b/src/runtime-state.ts new file mode 100644 index 0000000..536811c --- /dev/null +++ b/src/runtime-state.ts @@ -0,0 +1,133 @@ +import fs from 'fs'; +import path from 'path'; + +import { listAvailableGroups } from './available-groups.js'; +import { + type AssignRoomInput, + assignRoom, + deleteAllSessionsForGroup, + deleteSession, + getAllRoomBindings, + getAllSessions, + getRouterState, + setRouterState, + setSession, +} from './db.js'; +import { resolveGroupFolderPath } from './group-folder.js'; +import { logger } from './logger.js'; +import { normalizeStoredSeqCursor } from './message-cursor.js'; +import type { AgentType, RegisteredGroup } from './types.js'; + +export interface RuntimeState { + loadState: () => void; + saveState: () => void; + clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; + assignRoomForIpc: (jid: string, input: AssignRoomInput) => void; + getAvailableGroups: () => import('./agent-runner.js').AvailableGroup[]; + getLastTimestamp: () => string; + setLastTimestamp: (timestamp: string) => void; + getLastAgentTimestamps: () => Record; + getSessions: () => Record; + persistSession: (groupFolder: string, sessionId: string) => void; + getRoomBindings: () => Record; + setRoomBindings: (groups: Record) => void; +} + +export function createRuntimeState(): RuntimeState { + let lastTimestamp = ''; + let sessions: Record = {}; + let roomBindings: Record = {}; + let lastAgentTimestamp: Record = {}; + + const saveState = (): void => { + setRouterState('last_seq', lastTimestamp); + setRouterState('last_agent_seq', JSON.stringify(lastAgentTimestamp)); + }; + + const clearSession = ( + groupFolder: string, + opts?: { allRoles?: boolean }, + ): void => { + delete sessions[groupFolder]; + if (opts?.allRoles) { + deleteAllSessionsForGroup(groupFolder); + } else { + deleteSession(groupFolder); + } + }; + + const persistSession = (groupFolder: string, sessionId: string): void => { + sessions[groupFolder] = sessionId; + setSession(groupFolder, sessionId); + }; + + const loadState = (): void => { + lastTimestamp = normalizeStoredSeqCursor(getRouterState('last_seq')); + const agentTs = getRouterState('last_agent_seq'); + try { + const parsed = agentTs + ? (JSON.parse(agentTs) as Record) + : {}; + lastAgentTimestamp = Object.fromEntries( + Object.entries(parsed).map(([chatJid, cursor]) => [ + chatJid, + normalizeStoredSeqCursor(cursor, chatJid), + ]), + ); + } catch { + logger.warn('Corrupted last_agent_seq in DB, resetting'); + lastAgentTimestamp = {}; + } + sessions = getAllSessions(); + roomBindings = getAllRoomBindings(); + logger.info( + { + groupCount: Object.keys(roomBindings).length, + agentType: 'unified' satisfies AgentType | 'unified', + }, + 'State loaded', + ); + }; + + const assignRoomForIpc = (jid: string, input: AssignRoomInput): void => { + const assignedGroup = assignRoom(jid, input); + if (!assignedGroup) { + logger.warn({ jid }, 'Failed to assign room from IPC'); + return; + } + + let groupDir: string; + try { + groupDir = resolveGroupFolderPath(assignedGroup.folder); + } catch (err) { + logger.warn( + { jid, folder: assignedGroup.folder, err }, + 'Rejecting room assignment with invalid folder', + ); + return; + } + + const { jid: _ignoredJid, ...storedGroup } = assignedGroup; + roomBindings[jid] = storedGroup; + fs.mkdirSync(path.join(groupDir, 'logs'), { recursive: true }); + }; + + return { + loadState, + saveState, + clearSession, + assignRoomForIpc, + getAvailableGroups: () => listAvailableGroups(roomBindings), + getLastTimestamp: () => lastTimestamp, + setLastTimestamp: (timestamp) => { + lastTimestamp = timestamp; + }, + getLastAgentTimestamps: () => lastAgentTimestamp, + getSessions: () => sessions, + persistSession, + getRoomBindings: () => roomBindings, + setRoomBindings: (groups) => { + roomBindings = groups; + }, + }; +}