fix: salvage EJClaw runtime hardening

- add dist freshness guard to deploy flow
- preserve message source provenance for IPC injection
- harden Codex usage fallback and workspace package manager detection
- document owner branch return protocol
This commit is contained in:
ejclaw
2026-04-24 15:02:09 +09:00
parent a36c8a5f07
commit e036521054
19 changed files with 1059 additions and 51 deletions

View File

@@ -11,7 +11,8 @@
"build:runners": "bun run install:runners && bun run --cwd runners/shared build && bun run --cwd runners/agent-runner build && bun run --cwd runners/codex-runner build",
"build:all": "bun run build && bun run build:runners",
"build:runtime": "bun run build:all",
"deploy": "git pull --ff-only && bun run build:all && bun setup/index.ts --step migrate-room-registrations && systemctl --user restart ejclaw",
"verify:dist": "bash scripts/check-dist-fresh.sh",
"deploy": "git pull --ff-only && bun run build:all && bun run verify:dist && bun setup/index.ts --step migrate-room-registrations && systemctl --user restart ejclaw",
"start": "bun dist/index.js",
"dev": "bun --watch src/index.ts",
"test": "vitest run",

View File

@@ -41,3 +41,34 @@ Challenge the reviewer's reasoning. Point out logical gaps, over-engineering, sc
- Implementation, commits, and pushes require agreement from both sides. Either can veto
- Implement directly when it makes sense — you have full implementation authority
- Never mention or tag the user (@username) during the owner↔reviewer loop — the system handles escalation automatically. User is only notified when all resolution paths (including arbiter) are exhausted
## 🔴 Workspace Branch Protocol (MANDATORY)
The owner workspace is managed by EJClaw's paired-room state machine. The workspace branch name MUST be `codex/owner/<group-folder>` at the moment your turn ends. If any other branch is checked out when the next message arrives, the entire room goes BLOCKED with "branch mismatch" and needs manual git recovery.
### Every turn, in order
1. **Start**: verify `git branch --show-current`. If it is not `codex/owner/<group-folder>`, check it out before doing anything else.
2. **Work**: feel free to create feature branches (`fix/...`, `feat/...`) while implementing.
3. **Before you finish the turn**:
- If you committed on a feature branch, merge it back into the owner branch (`git checkout codex/owner/<group-folder> && git merge --ff-only <feature-branch>`), then optionally delete the feature branch.
- If the feature branch is behind or diverged, use the appropriate merge/rebase to land the work on the owner branch.
- Confirm a clean end-state: `git branch --show-current` prints `codex/owner/<group-folder>` and `git status --short` is empty (or at least there is no merge conflict and no stray dirty state that the next turn cannot understand).
4. **Only then** emit your DONE / DONE_WITH_CONCERNS / BLOCKED / NEEDS_CONTEXT line and exit.
### Hard rules
- Never end a turn on a `fix/*` or `feat/*` branch.
- Never end a turn with unresolved merge conflicts.
- Never leave the workspace in detached-HEAD or rebase-in-progress state at turn end.
### If you discover a mismatch at turn start
You landed on a feature branch because a previous turn forgot to return. Recover and continue:
1. `git status` — inspect dirty state.
2. If clean and the feature branch is ahead of the owner branch with a linear history: `git checkout codex/owner/<group-folder> && git merge --ff-only <feature-branch>`.
3. If there is dirty state you actually want to keep: commit it with a meaningful message or `git stash push -u -m "<reason>"`, then switch branches and re-apply.
4. If the feature branch has diverged in a way that is not fast-forwardable: branch it off explicitly (`backup/<group>-recover-<timestamp>`), return to the owner branch, then merge/cherry-pick the needed commits.
The group folder matches the EJClaw paired-room workspace directory name (for example `eyejokerdb-9`, `ejset`, `brain`). Use `basename $(pwd | sed 's|/owner$||')` if in doubt.

113
scripts/check-dist-fresh.sh Executable file
View File

@@ -0,0 +1,113 @@
#!/usr/bin/env bash
# check-dist-fresh.sh — verify compiled dist/ outputs are at least as new as their
# corresponding src/ .ts sources. Use this as a restart-precondition / CI guard so
# we never ship stale compiled artifacts again (see 2026-04-23 reviewer infinite-loop
# incident caused by a patched src/bundled-cli-path.ts with a stale dist/).
#
# Usage:
# scripts/check-dist-fresh.sh # checks default packages (root + agent-runner)
# scripts/check-dist-fresh.sh DIR [DIR ...] # each DIR must contain src/ and dist/
#
# Exit codes:
# 0 — all dist/ files up-to-date
# 1 — at least one dist/ file missing or older than its src/ sibling
# 2 — usage / environment error
#
# Rules per package:
# - For every foo.ts under src/ (excluding *.test.ts, *.d.ts, __tests__/*):
# require dist/foo.js to exist AND have mtime >= src/foo.ts
# - Missing dist/foo.js is a hard error (build never ran for that file).
#
set -euo pipefail
# Resolve repo root from this script's location so the script works from any cwd.
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd -- "${SCRIPT_DIR}/.." && pwd)"
if [ "$#" -gt 0 ]; then
PACKAGES=("$@")
else
PACKAGES=(
"${REPO_ROOT}"
"${REPO_ROOT}/runners/agent-runner"
"${REPO_ROOT}/runners/codex-runner"
"${REPO_ROOT}/runners/shared"
)
fi
fail=0
checked=0
# Use mtime in nanoseconds (GNU stat) for precision. Fallback to seconds if %Y only.
_mtime() {
# $1: path
local t
t="$(stat -c '%Y' "$1" 2>/dev/null || true)"
if [ -z "$t" ]; then
# BSD stat fallback (macOS)
t="$(stat -f '%m' "$1" 2>/dev/null || true)"
fi
echo "${t:-0}"
}
check_package() {
local pkg="$1"
local src_dir="${pkg}/src"
local dist_dir="${pkg}/dist"
if [ ! -d "$src_dir" ]; then
echo "skip: ${pkg} (no src/)" >&2
return 0
fi
if [ ! -d "$dist_dir" ]; then
echo "FAIL: ${pkg} — dist/ does not exist (run build)" >&2
fail=1
return 0
fi
local ts dist_file src_mtime dist_mtime
# Loop over every .ts file under src/, skipping tests & type declarations.
while IFS= read -r -d '' ts; do
local rel="${ts#${src_dir}/}"
dist_file="${dist_dir}/${rel%.ts}.js"
checked=$((checked + 1))
if [ ! -f "$dist_file" ]; then
echo "FAIL: missing dist file: ${dist_file} (src: ${ts})" >&2
fail=1
continue
fi
src_mtime="$(_mtime "$ts")"
dist_mtime="$(_mtime "$dist_file")"
if [ "$dist_mtime" -lt "$src_mtime" ]; then
echo "FAIL: stale dist: ${dist_file}" >&2
echo " src mtime=${src_mtime} ($(date -d "@${src_mtime}" '+%F %T' 2>/dev/null || true)) ${ts}" >&2
echo " dist mtime=${dist_mtime} ($(date -d "@${dist_mtime}" '+%F %T' 2>/dev/null || true))" >&2
fail=1
fi
done < <(find "$src_dir" -type f -name '*.ts' \
! -name '*.test.ts' \
! -name '*.d.ts' \
! -path '*/__tests__/*' \
-print0)
}
for pkg in "${PACKAGES[@]}"; do
if [ ! -d "$pkg" ]; then
echo "skip: ${pkg} (directory not found)" >&2
continue
fi
check_package "$pkg"
done
if [ "$fail" -ne 0 ]; then
echo "" >&2
echo "dist freshness check FAILED (checked ${checked} source files)" >&2
echo "Run: bun run build:all (or the package's build script)" >&2
exit 1
fi
echo "dist freshness OK (checked ${checked} source files across ${#PACKAGES[@]} package(s))"
exit 0

View File

@@ -56,6 +56,19 @@ function createFakeAccounts(homeDir: string, count: number): void {
}
}
function createDefaultCodexAuth(homeDir: string): string {
const authPath = path.join(homeDir, '.codex', 'auth.json');
fs.mkdirSync(path.dirname(authPath), { recursive: true });
fs.writeFileSync(
authPath,
JSON.stringify({
auth_mode: 'chatgpt',
tokens: { account_id: 'default-acct', access_token: 'default-token' },
}),
);
return authPath;
}
describe('codex-token-rotation d7 ≥ 100% auto-skip', () => {
let tempHome: string;
@@ -142,4 +155,46 @@ describe('codex-token-rotation d7 ≥ 100% auto-skip', () => {
'Failed to persist Codex rotation state',
);
});
it('does not append ~/.codex/auth.json fallback when numbered accounts exist', async () => {
const fallbackAuthPath = createDefaultCodexAuth(tempHome);
const mod = await import('./codex-token-rotation.js');
mod.initCodexTokenRotation();
expect(mod.getCodexAccountCount()).toBe(4);
expect(mod.getActiveCodexAuthPath()).not.toBe(fallbackAuthPath);
});
});
describe('codex-token-rotation single-account fallback', () => {
let tempHome: string;
beforeEach(() => {
vi.resetModules();
tempHome = fs.mkdtempSync(path.join('/tmp', 'ejclaw-codex-fallback-'));
process.env.CODEX_ROT_TEST_HOME = tempHome;
});
afterEach(() => {
delete process.env.CODEX_ROT_TEST_HOME;
fs.rmSync(tempHome, { recursive: true, force: true });
});
it('uses ~/.codex/auth.json fallback when ~/.codex-accounts is absent', async () => {
const fallbackAuthPath = createDefaultCodexAuth(tempHome);
const mod = await import('./codex-token-rotation.js');
mod.initCodexTokenRotation();
expect(mod.getCodexAccountCount()).toBe(1);
expect(mod.getActiveCodexAuthPath()).toBe(fallbackAuthPath);
expect(mod.getAllCodexAccounts()).toEqual([
expect.objectContaining({
index: 0,
accountId: 'default-acct',
isActive: true,
}),
]);
});
});

View File

@@ -78,46 +78,63 @@ let currentIndex = 0;
let initialized = false;
const ACCOUNTS_DIR = path.join(os.homedir(), '.codex-accounts');
const DEFAULT_CODEX_DIR = path.join(os.homedir(), '.codex');
const DEFAULT_AUTH_PATH = path.join(DEFAULT_CODEX_DIR, 'auth.json');
function loadCodexAccount(
authPath: string,
fallbackAccountId: string,
): boolean {
const data = readJsonFile<{
tokens?: { account_id?: string; id_token?: string };
}>(authPath);
if (!data) {
logger.warn({ authPath }, 'Failed to parse codex account auth.json');
return false;
}
const accountId = data?.tokens?.account_id || fallbackAccountId;
const jwt = parseJwtAuth(data?.tokens?.id_token || '');
accounts.push({
index: accounts.length,
authPath,
accountId,
planType: jwt.planType,
subscriptionUntil: jwt.expiresAt,
rateLimitedUntil: null,
});
return true;
}
export function initCodexTokenRotation(): void {
if (initialized) return;
initialized = true;
if (!fs.existsSync(ACCOUNTS_DIR)) {
logger.info(
{ dir: ACCOUNTS_DIR },
'Codex accounts dir not found, skipping',
);
return;
}
const hasAccountsDir = fs.existsSync(ACCOUNTS_DIR);
const dirs = hasAccountsDir
? fs
.readdirSync(ACCOUNTS_DIR)
.filter((d) => /^\d+$/.test(d))
.sort((a, b) => parseInt(a) - parseInt(b))
: [];
const dirs = fs
.readdirSync(ACCOUNTS_DIR)
.filter((d) => /^\d+$/.test(d))
.sort((a, b) => parseInt(a) - parseInt(b));
if (!hasAccountsDir) {
logger.info({ dir: ACCOUNTS_DIR }, 'Codex accounts dir not found');
}
for (const dir of dirs) {
const authPath = path.join(ACCOUNTS_DIR, dir, 'auth.json');
if (!fs.existsSync(authPath)) continue;
loadCodexAccount(authPath, `account-${dir}`);
}
const data = readJsonFile<{
tokens?: { account_id?: string; id_token?: string };
}>(authPath);
if (!data) {
logger.warn({ authPath }, 'Failed to parse codex account auth.json');
continue;
if (dirs.length === 0 && fs.existsSync(DEFAULT_AUTH_PATH)) {
if (loadCodexAccount(DEFAULT_AUTH_PATH, 'default-account')) {
logger.info(
{ authPath: DEFAULT_AUTH_PATH },
'Codex accounts dir absent/empty; using ~/.codex/auth.json fallback',
);
}
const accountId = data?.tokens?.account_id || `account-${dir}`;
const jwt = parseJwtAuth(data?.tokens?.id_token || '');
const planType = jwt.planType;
accounts.push({
index: accounts.length,
authPath,
accountId,
planType,
subscriptionUntil: jwt.expiresAt,
rateLimitedUntil: null,
});
}
if (accounts.length > 1) loadCodexState();
@@ -236,8 +253,14 @@ export function reloadCodexStateFromDisk(): void {
/** Get the auth.json path for the current active account. */
export function getActiveCodexAuthPath(): string | null {
return getCodexAuthPath();
}
export function getCodexAuthPath(
accountIndex: number = currentIndex,
): string | null {
if (accounts.length === 0) return null;
return accounts[currentIndex]?.authPath ?? null;
return accounts[accountIndex]?.authPath ?? null;
}
export function detectCodexRotationTrigger(

View File

@@ -0,0 +1,242 @@
import { EventEmitter } from 'events';
import fs from 'fs';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('child_process', () => ({
spawn: vi.fn(),
}));
vi.mock('./logger.js', () => ({
logger: {
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
debug: vi.fn(),
},
}));
vi.mock('./config.js', () => ({
DATA_DIR: '/tmp/ejclaw-codex-usage-data',
}));
vi.mock('./utils.js', async () => {
const actual =
await vi.importActual<typeof import('./utils.js')>('./utils.js');
return {
...actual,
writeJsonFile: vi.fn(),
};
});
vi.mock('os', async () => {
const actual = await vi.importActual<typeof import('os')>('os');
return {
...actual,
default: {
...actual,
homedir: () => process.env.CODEX_USAGE_TEST_HOME || '/tmp',
},
homedir: () => process.env.CODEX_USAGE_TEST_HOME || '/tmp',
};
});
function createDefaultCodexAuth(homeDir: string): string {
const authPath = path.join(homeDir, '.codex', 'auth.json');
fs.mkdirSync(path.dirname(authPath), { recursive: true });
fs.writeFileSync(
authPath,
JSON.stringify({
auth_mode: 'chatgpt',
tokens: { account_id: 'default-acct', access_token: 'default-token' },
}),
);
return authPath;
}
function createFakeChildProcess(rateLimitsByLimitId: Record<string, unknown>) {
const proc = new EventEmitter() as EventEmitter & {
stdout: EventEmitter;
stdin: { write: ReturnType<typeof vi.fn> };
kill: ReturnType<typeof vi.fn>;
};
const stdout = new EventEmitter();
proc.stdout = stdout;
proc.stdin = {
write: vi.fn((payload: string) => {
const message = JSON.parse(payload.trim()) as {
id: number;
method?: string;
};
if (message.id === 1) {
setImmediate(() => {
stdout.emit(
'data',
Buffer.from(`${JSON.stringify({ id: 1, result: {} })}\n`),
);
});
}
if (message.id === 2 && message.method === 'account/rateLimits/read') {
setImmediate(() => {
stdout.emit(
'data',
Buffer.from(
`${JSON.stringify({
id: 2,
result: { rateLimitsByLimitId },
})}\n`,
),
);
});
}
return true;
}),
};
proc.kill = vi.fn();
return proc;
}
describe('codex-usage-collector fallback account usage', () => {
let tempHome: string;
beforeEach(() => {
vi.resetModules();
vi.clearAllMocks();
tempHome = fs.mkdtempSync(path.join('/tmp', 'ejclaw-codex-usage-'));
process.env.CODEX_USAGE_TEST_HOME = tempHome;
});
afterEach(() => {
delete process.env.CODEX_USAGE_TEST_HOME;
fs.rmSync(tempHome, { recursive: true, force: true });
});
it('refreshes usage via ~/.codex/auth.json when ~/.codex-accounts is missing', async () => {
createDefaultCodexAuth(tempHome);
const childProcess = await import('child_process');
const fallbackCodexHome = path.join(tempHome, '.codex');
let capturedCodexHome: string | undefined;
vi.mocked(childProcess.spawn).mockImplementation(((
_cmd: string,
_args: readonly string[] | undefined,
opts?: { env?: Record<string, string> },
) => {
capturedCodexHome = opts?.env?.CODEX_HOME;
return createFakeChildProcess({
codex: {
limitName: 'Codex',
primary: {
usedPercent: 12.4,
resetsAt: new Date(Date.now() + 3_600_000).toISOString(),
},
secondary: {
usedPercent: 67.6,
resetsAt: new Date(Date.now() + 86_400_000).toISOString(),
},
},
}) as never;
}) as unknown as typeof childProcess.spawn);
const rotation = await import('./codex-token-rotation.js');
const usage = await import('./codex-usage-collector.js');
rotation.initCodexTokenRotation();
const result = await usage.refreshActiveCodexUsage();
expect(rotation.getCodexAccountCount()).toBe(1);
expect(capturedCodexHome).toBe(fallbackCodexHome);
expect(result.fetchedAt).toEqual(expect.any(String));
expect(result.rows).toEqual([
expect.objectContaining({
name: 'Codex',
h5pct: 12,
d7pct: 68,
}),
]);
});
it('finds codex via ~/.hermes/node/bin when running under bun', async () => {
createDefaultCodexAuth(tempHome);
const childProcess = await import('child_process');
const bunExecPath = path.join(tempHome, '.bun', 'bin', 'bun');
const hermesBin = path.join(tempHome, '.hermes', 'node', 'bin');
const originalExecPath = process.execPath;
const originalPath = process.env.PATH;
fs.mkdirSync(hermesBin, { recursive: true });
process.env.PATH = '/usr/local/bin:/usr/bin:/bin';
Object.defineProperty(process, 'execPath', {
configurable: true,
writable: true,
value: bunExecPath,
});
vi.mocked(childProcess.spawn).mockImplementation(((
cmd: string,
_args: readonly string[] | undefined,
opts?: { env?: Record<string, string> },
) => {
const spawnPathEntries = (opts?.env?.PATH || '').split(path.delimiter);
expect(cmd).toBe('codex');
expect(spawnPathEntries).toContain(path.dirname(bunExecPath));
expect(spawnPathEntries).toContain(hermesBin);
if (!spawnPathEntries.includes(hermesBin)) {
const proc = new EventEmitter() as EventEmitter & {
stdout: EventEmitter;
stdin: { write: ReturnType<typeof vi.fn> };
kill: ReturnType<typeof vi.fn>;
};
proc.stdout = new EventEmitter();
proc.stdin = { write: vi.fn() };
proc.kill = vi.fn();
setImmediate(() => proc.emit('error', new Error('spawn codex ENOENT')));
return proc as never;
}
return createFakeChildProcess({
codex: {
limitName: 'Codex',
primary: {
usedPercent: 34.2,
resetsAt: new Date(Date.now() + 7_200_000).toISOString(),
},
secondary: {
usedPercent: 56.1,
resetsAt: new Date(Date.now() + 172_800_000).toISOString(),
},
},
}) as never;
}) as unknown as typeof childProcess.spawn);
try {
const rotation = await import('./codex-token-rotation.js');
const usage = await import('./codex-usage-collector.js');
rotation.initCodexTokenRotation();
const result = await usage.refreshActiveCodexUsage();
expect(result.fetchedAt).toEqual(expect.any(String));
expect(result.rows).toEqual([
expect.objectContaining({
name: 'Codex',
h5pct: 34,
d7pct: 56,
}),
]);
} finally {
Object.defineProperty(process, 'execPath', {
configurable: true,
writable: true,
value: originalExecPath,
});
if (originalPath === undefined) {
delete process.env.PATH;
} else {
process.env.PATH = originalPath;
}
}
});
});

View File

@@ -5,6 +5,7 @@ import path from 'path';
import {
getAllCodexAccounts,
getCodexAuthPath,
updateCodexAccountUsage,
} from './codex-token-rotation.js';
import { formatResetRemaining, type UsageRow } from './dashboard-usage-rows.js';
@@ -27,11 +28,26 @@ export interface CodexUsageRefreshResult {
fetchedAt: string | null;
}
const CODEX_ACCOUNTS_DIR = path.join(os.homedir(), '.codex-accounts');
/** Full scan interval — exported so the orchestrator can schedule it. */
export const CODEX_FULL_SCAN_INTERVAL = 3_600_000; // 1 hour
function getPreferredCodexPathEntries(): string[] {
const entries = [
path.dirname(process.execPath),
path.join(os.homedir(), '.npm-global', 'bin'),
];
if (process.versions.bun || path.basename(process.execPath) === 'bun') {
entries.push(path.join(os.homedir(), '.hermes', 'node', 'bin'));
}
return [...new Set(entries)];
}
function getCodexHomeForAccount(accountIndex?: number): string | null {
const authPath = getCodexAuthPath(accountIndex);
if (!authPath || !fs.existsSync(authPath)) return null;
return path.dirname(authPath);
}
export async function fetchCodexUsage(
codexHomeOverride?: string,
): Promise<CodexRateLimit[] | null> {
@@ -59,11 +75,9 @@ export async function fetchCodexUsage(
const spawnEnv: Record<string, string> = {
...(process.env as Record<string, string>),
PATH: [
path.dirname(process.execPath),
path.join(os.homedir(), '.npm-global', 'bin'),
process.env.PATH || '',
].join(':'),
PATH: [...getPreferredCodexPathEntries(), process.env.PATH || '']
.filter(Boolean)
.join(path.delimiter),
};
if (codexHomeOverride) {
spawnEnv.CODEX_HOME = codexHomeOverride;
@@ -243,8 +257,8 @@ export async function refreshAllCodexAccountUsage(): Promise<CodexUsageRefreshRe
let anySuccess = false;
for (const acct of codexAccounts) {
const accountDir = path.join(CODEX_ACCOUNTS_DIR, String(acct.index + 1));
if (!fs.existsSync(accountDir)) continue;
const accountDir = getCodexHomeForAccount(acct.index);
if (!accountDir) continue;
try {
const usage = await fetchCodexUsage(accountDir);
@@ -281,8 +295,8 @@ export async function refreshActiveCodexUsage(): Promise<CodexUsageRefreshResult
return { rows: buildCodexUsageRowsFromState(), fetchedAt: null };
}
const accountDir = path.join(CODEX_ACCOUNTS_DIR, String(active.index + 1));
if (!fs.existsSync(accountDir)) {
const accountDir = getCodexHomeForAccount(active.index);
if (!accountDir) {
return { rows: buildCodexUsageRowsFromState(), fetchedAt: null };
}

View File

@@ -0,0 +1,157 @@
import { Database } from 'bun:sqlite';
import fs from 'fs';
import os from 'os';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import {
_initTestDatabase,
_initTestDatabaseFromFile,
getMessagesSinceSeq,
getRecentChatMessages,
storeChatMetadata,
storeMessage,
} from './db.js';
describe('message_source_kind persistence', () => {
let tempDir: string | null = null;
beforeEach(() => {
_initTestDatabase();
});
afterEach(() => {
if (tempDir) {
fs.rmSync(tempDir, { recursive: true, force: true });
tempDir = null;
}
});
it('persists explicit message source kind through store and read paths', () => {
storeChatMetadata('room@g.us', '2026-04-24T00:00:00.000Z', 'Room');
storeMessage({
id: 'ipc-1',
chat_jid: 'room@g.us',
sender: 'hermes',
sender_name: 'Hermes',
content: 'wake up',
timestamp: '2026-04-24T00:00:01.000Z',
is_from_me: false,
is_bot_message: false,
message_source_kind: 'trusted_external_bot',
});
const [recent] = getRecentChatMessages('room@g.us', 1);
expect(recent).toMatchObject({
id: 'ipc-1',
is_bot_message: false,
message_source_kind: 'trusted_external_bot',
});
storeMessage({
id: 'ipc-1',
chat_jid: 'room@g.us',
sender: 'hermes',
sender_name: 'Hermes',
content: 'updated',
timestamp: '2026-04-24T00:00:02.000Z',
is_from_me: false,
is_bot_message: true,
message_source_kind: 'ipc_injected_bot',
});
const [updated] = getMessagesSinceSeq('room@g.us', 0, 'EJClaw', 10);
expect(updated).toMatchObject({
id: 'ipc-1',
content: 'updated',
is_bot_message: true,
message_source_kind: 'ipc_injected_bot',
});
});
it('defaults missing message source kind from is_bot_message', () => {
storeChatMetadata('room@g.us', '2026-04-24T00:00:00.000Z', 'Room');
storeMessage({
id: 'human-1',
chat_jid: 'room@g.us',
sender: 'user',
sender_name: 'User',
content: 'hello',
timestamp: '2026-04-24T00:00:01.000Z',
is_from_me: false,
is_bot_message: false,
});
storeMessage({
id: 'bot-1',
chat_jid: 'room@g.us',
sender: 'bot',
sender_name: 'Bot',
content: 'done',
timestamp: '2026-04-24T00:00:02.000Z',
is_from_me: false,
is_bot_message: true,
});
expect(getRecentChatMessages('room@g.us', 2)).toEqual([
expect.objectContaining({ id: 'human-1', message_source_kind: 'human' }),
expect.objectContaining({ id: 'bot-1', message_source_kind: 'bot' }),
]);
});
it('migrates legacy message tables and backfills source kind', () => {
tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-msg-source-'));
const dbPath = path.join(tempDir, 'messages.db');
const legacyDb = new Database(dbPath);
legacyDb.exec(`
CREATE TABLE chats (
jid TEXT PRIMARY KEY,
name TEXT,
last_message_time TEXT,
channel TEXT,
is_group INTEGER DEFAULT 0
);
CREATE TABLE messages (
id TEXT,
chat_jid TEXT,
sender TEXT,
sender_name TEXT,
content TEXT,
timestamp TEXT,
seq INTEGER,
is_from_me INTEGER,
is_bot_message INTEGER DEFAULT 0,
PRIMARY KEY (id, chat_jid)
);
CREATE TABLE schema_migrations (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
);
INSERT INTO chats (jid, name, last_message_time, channel, is_group)
VALUES ('room@g.us', 'Room', '2026-04-24T00:00:02.000Z', 'discord', 1);
INSERT INTO messages (
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
) VALUES
('human-legacy', 'room@g.us', 'user', 'User', 'hello', '2026-04-24T00:00:01.000Z', 1, 0, 0),
('bot-legacy', 'room@g.us', 'bot', 'Bot', 'done', '2026-04-24T00:00:02.000Z', 2, 0, 1);
`);
const insertMigration = legacyDb.prepare(
'INSERT INTO schema_migrations (version, name) VALUES (?, ?)',
);
for (let version = 1; version <= 12; version += 1) {
insertMigration.run(version, `legacy-${version}`);
}
legacyDb.close();
_initTestDatabaseFromFile(dbPath);
expect(getRecentChatMessages('room@g.us', 2)).toEqual([
expect.objectContaining({
id: 'human-legacy',
message_source_kind: 'human',
}),
expect.objectContaining({ id: 'bot-legacy', message_source_kind: 'bot' }),
]);
});
});

View File

@@ -19,6 +19,7 @@ export function applyBaseSchema(database: Database): void {
seq INTEGER,
is_from_me INTEGER,
is_bot_message INTEGER DEFAULT 0,
message_source_kind TEXT NOT NULL DEFAULT 'human',
PRIMARY KEY (id, chat_jid),
FOREIGN KEY (chat_jid) REFERENCES chats(jid)
);

View File

@@ -1,5 +1,9 @@
import { Database } from 'bun:sqlite';
import {
inferMessageSourceKindFromBotFlag,
normalizeMessageSourceKind,
} from '../message-source.js';
import { NewMessage } from '../types.js';
export interface ChatInfo {
@@ -14,12 +18,18 @@ function normalizeMessageRow(
row: NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
message_source_kind?: unknown;
},
): NewMessage {
const isBotMessage = !!row.is_bot_message;
return {
...row,
is_from_me: !!row.is_from_me,
is_bot_message: !!row.is_bot_message,
is_bot_message: isBotMessage,
message_source_kind: normalizeMessageSourceKind(
row.message_source_kind,
inferMessageSourceKindFromBotFlag(isBotMessage),
),
};
}
@@ -106,15 +116,16 @@ export function storeMessageInDatabase(
database
.prepare(
`INSERT INTO messages (
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id, chat_jid) DO UPDATE SET
sender = excluded.sender,
sender_name = excluded.sender_name,
content = excluded.content,
timestamp = excluded.timestamp,
is_from_me = excluded.is_from_me,
is_bot_message = excluded.is_bot_message`,
is_bot_message = excluded.is_bot_message,
message_source_kind = excluded.message_source_kind`,
)
.run(
msg.id,
@@ -126,6 +137,10 @@ export function storeMessageInDatabase(
seq,
msg.is_from_me ? 1 : 0,
msg.is_bot_message ? 1 : 0,
normalizeMessageSourceKind(
msg.message_source_kind,
inferMessageSourceKindFromBotFlag(msg.is_bot_message),
),
);
})();
}
@@ -142,7 +157,7 @@ export function getNewMessagesFromDatabase(
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE timestamp > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
@@ -178,7 +193,7 @@ export function getMessagesSinceFromDatabase(
): NewMessage[] {
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ? AND timestamp > ?
AND content NOT LIKE ?
@@ -238,7 +253,7 @@ export function getNewMessagesBySeqFromDatabase(
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE seq > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
@@ -273,7 +288,7 @@ export function getMessagesSinceSeqFromDatabase(
): NewMessage[] {
const sinceSeq = normalizeSeqCursor(sinceSeqCursor);
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ? AND seq > ?
AND content NOT LIKE ?
@@ -300,7 +315,7 @@ export function getRecentChatMessagesFromDatabase(
): NewMessage[] {
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ?
AND content != '' AND content IS NOT NULL

View File

@@ -0,0 +1,43 @@
import type { Database } from 'bun:sqlite';
import { tableHasColumn } from './helpers.js';
import type { SchemaMigrationDefinition } from './types.js';
export const MESSAGE_SOURCE_KIND_MIGRATION: SchemaMigrationDefinition = {
version: 13,
name: 'message_source_kind',
apply(database: Database) {
const addedColumn = !tableHasColumn(
database,
'messages',
'message_source_kind',
);
if (addedColumn) {
database.exec(`
ALTER TABLE messages
ADD COLUMN message_source_kind TEXT NOT NULL DEFAULT 'human'
`);
}
const invalidOnlyWhere = `
WHERE message_source_kind IS NULL
OR message_source_kind = ''
OR message_source_kind NOT IN (
'human',
'bot',
'trusted_external_bot',
'ipc_injected_human',
'ipc_injected_bot'
)
`;
database.exec(`
UPDATE messages
SET message_source_kind = CASE
WHEN is_bot_message = 1 THEN 'bot'
ELSE 'human'
END
${addedColumn ? '' : invalidOnlyWhere}
`);
},
};

View File

@@ -12,6 +12,7 @@ import { PAIRED_WORKSPACE_PROJECT_SCHEMA_CLEANUP_MIGRATION } from './009_paired-
import { PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION } from './010_paired-turn-provenance-upgrade.js';
import { OWNER_FAILURE_COUNT_MIGRATION } from './011_owner-failure-count.js';
import { PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION } from './012_paired-verdict-and-step-telemetry.js';
import { MESSAGE_SOURCE_KIND_MIGRATION } from './013_message-source-kind.js';
import type {
SchemaMigrationArgs,
SchemaMigrationDefinition,
@@ -32,6 +33,7 @@ const ORDERED_SCHEMA_MIGRATIONS: readonly SchemaMigrationDefinition[] = [
PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION,
OWNER_FAILURE_COUNT_MIGRATION,
PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION,
MESSAGE_SOURCE_KIND_MIGRATION,
];
function ensureSchemaMigrationsTable(database: Database): void {

View File

@@ -61,6 +61,10 @@ import {
hasAvailableClaudeToken,
initTokenRotation,
} from './token-rotation.js';
import {
isBotMessageSourceKind,
resolveInjectedMessageSourceKind,
} from './message-source.js';
import { parseVisibleVerdict } from './paired-execution-context-shared.js';
export function isTerminalStatusMessage(text: string): boolean {
@@ -412,6 +416,51 @@ async function main(): Promise<void> {
queue.noteDirectTerminalDelivery(jid, senderRole, text);
}
},
injectInboundMessage: async (payload) => {
const jid = payload.chatJid;
const binding = runtimeState.getRoomBindings()[jid];
if (!binding) {
logger.warn(
{ chatJid: jid, sender: payload.sender ?? null },
'inject_inbound_message: no room binding, dropping',
);
return;
}
const ts = payload.timestamp || new Date().toISOString();
const msgId =
payload.messageId ||
`ipc-inject-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const treatAsHuman = payload.treatAsHuman === true;
const messageSourceKind = resolveInjectedMessageSourceKind({
treatAsHuman,
sourceKind: payload.sourceKind,
});
storeChatMetadata(jid, ts, binding.name, 'discord', true);
storeMessage({
id: msgId,
chat_jid: jid,
sender: payload.sender || 'ipc-inject',
sender_name: payload.senderName || payload.sender || 'IPC Inject',
content: payload.text,
timestamp: ts,
is_from_me: false,
is_bot_message: isBotMessageSourceKind(messageSourceKind),
message_source_kind: messageSourceKind,
});
queue.enqueueMessageCheck(jid, resolveGroupIpcPath(binding.folder));
logger.info(
{
chatJid: jid,
sender: payload.sender ?? null,
senderName: payload.senderName ?? null,
treatAsHuman,
messageSourceKind,
messageId: msgId,
groupFolder: binding.folder,
},
'Injected inbound message via IPC',
);
},
nudgeScheduler: nudgeSchedulerLoop,
roomBindings: runtimeState.getRoomBindings,
assignRoom: runtimeState.assignRoomForIpc,

View File

@@ -0,0 +1,43 @@
import { describe, expect, it } from 'vitest';
import {
isBotMessageSourceKind,
normalizeMessageSourceKind,
resolveInjectedMessageSourceKind,
} from './message-source.js';
describe('message source helpers', () => {
it('defaults IPC injected messages to trusted human-equivalent provenance', () => {
expect(resolveInjectedMessageSourceKind({ treatAsHuman: true })).toBe(
'trusted_external_bot',
);
expect(
isBotMessageSourceKind(
resolveInjectedMessageSourceKind({ treatAsHuman: true }),
),
).toBe(false);
});
it('defaults non-human IPC injected messages to bot-equivalent provenance', () => {
expect(resolveInjectedMessageSourceKind({ treatAsHuman: false })).toBe(
'ipc_injected_bot',
);
expect(
isBotMessageSourceKind(
resolveInjectedMessageSourceKind({ treatAsHuman: false }),
),
).toBe(true);
});
it('honors valid explicit source kinds and normalizes invalid values', () => {
expect(
resolveInjectedMessageSourceKind({
treatAsHuman: true,
sourceKind: 'ipc_injected_human',
}),
).toBe('ipc_injected_human');
expect(normalizeMessageSourceKind('bot', 'human')).toBe('bot');
expect(normalizeMessageSourceKind('not-real', 'human')).toBe('human');
});
});

39
src/message-source.ts Normal file
View File

@@ -0,0 +1,39 @@
import type { MessageSourceKind } from './types.js';
const MESSAGE_SOURCE_KINDS = new Set<MessageSourceKind>([
'human',
'bot',
'trusted_external_bot',
'ipc_injected_human',
'ipc_injected_bot',
]);
export function normalizeMessageSourceKind(
value: unknown,
fallback: MessageSourceKind = 'human',
): MessageSourceKind {
return typeof value === 'string' &&
MESSAGE_SOURCE_KINDS.has(value as MessageSourceKind)
? (value as MessageSourceKind)
: fallback;
}
export function isBotMessageSourceKind(kind: MessageSourceKind): boolean {
return kind === 'bot' || kind === 'ipc_injected_bot';
}
export function inferMessageSourceKindFromBotFlag(
isBotMessage: boolean | number | null | undefined,
): MessageSourceKind {
return isBotMessage ? 'bot' : 'human';
}
export function resolveInjectedMessageSourceKind(args: {
treatAsHuman: boolean;
sourceKind?: unknown;
}): MessageSourceKind {
return normalizeMessageSourceKind(
args.sourceKind,
args.treatAsHuman ? 'trusted_external_bot' : 'ipc_injected_bot',
);
}

View File

@@ -106,6 +106,66 @@ describe('verification helpers', () => {
});
});
it('verifies nested pnpm workspaces under a bun parent without tripping corepack project specs', async () => {
const parentDir = fs.mkdtempSync(
path.join(os.tmpdir(), 'ejclaw-verification-corepack-parent-'),
);
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'owner');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({
name: 'nested-pnpm-workspace',
scripts: {
typecheck:
'node -e "process.stdout.write(process.env.COREPACK_ENABLE_PROJECT_SPEC || \'missing\')"',
},
}),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
[
"lockfileVersion: '6.0'",
'settings:',
' autoInstallPeers: true',
' excludeLinksFromLockfile: false',
'importers:',
' .: {}',
'',
].join('\n'),
);
fs.mkdirSync(path.join(repoDir, 'node_modules', '.bin'), {
recursive: true,
});
fs.writeFileSync(
path.join(repoDir, 'node_modules', '.bin', 'placeholder'),
'',
);
expect(hasInstalledNodeModules(repoDir)).toBe(false);
const expectedSnapshotId = computeVerificationSnapshot(repoDir).snapshotId;
const result = await runVerificationRequest(
{
requestId: 'req-corepack-parent-pnpm',
profile: 'typecheck',
expectedSnapshotId,
},
{ repoDir },
);
expect(result.ok).toBe(true);
expect(result.exitCode).toBe(0);
expect(result.command).toBe('corepack pnpm run typecheck');
expect(result.stdout).toContain('0');
expect(result.snapshotId).toBe(expectedSnapshotId);
});
it('computes a stable snapshot over the readable workspace inputs', () => {
const repoDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-snapshot-'));
fs.mkdirSync(path.join(repoDir, 'src'), { recursive: true });

View File

@@ -5,9 +5,11 @@ import path from 'path';
import { TIMEZONE } from './config.js';
import {
buildWorkspaceCommandEnvironment,
buildWorkspaceScriptCommand,
ensureWorkspaceDependenciesInstalled,
hasInstalledNodeModules,
type WorkspacePackageManager,
} from './workspace-package-manager.js';
import { computeVerificationSnapshotId } from '../shared/verification-snapshot.js';
@@ -44,6 +46,7 @@ export interface VerificationSnapshot {
}
interface VerificationCommandSpec {
packageManager: WorkspacePackageManager;
file: string;
args: string[];
commandText: string;
@@ -111,6 +114,7 @@ export function buildVerificationCommand(
: 'lint';
const command = buildWorkspaceScriptCommand(repoDir, scriptName);
return {
packageManager: command.packageManager,
file: command.file,
args: command.args,
commandText: command.commandText,
@@ -334,7 +338,11 @@ export async function runVerificationRequest(
command.args,
{
cwd: repoDir,
env: directExecution.env,
env: buildWorkspaceCommandEnvironment(
repoDir,
command.packageManager,
directExecution.env,
),
},
);
const afterSnapshot = computeVerificationSnapshot(repoDir);

View File

@@ -13,6 +13,7 @@ vi.mock('child_process', () => ({
}));
import {
buildWorkspaceCommandEnvironment,
buildWorkspaceScriptCommand,
detectWorkspacePackageManager,
ensureWorkspaceDependenciesInstalled,
@@ -139,6 +140,13 @@ describe('workspace package manager helpers', () => {
packageManager: 'pnpm',
commandText: 'corepack pnpm install --frozen-lockfile',
});
expect(execFileSyncMock).toHaveBeenCalledWith(
'corepack',
['pnpm', 'install', '--frozen-lockfile'],
expect.objectContaining({
cwd: repoDir,
}),
);
expect(hasInstalledNodeModules(repoDir)).toBe(true);
execFileSyncMock.mockClear();
@@ -209,4 +217,59 @@ describe('workspace package manager helpers', () => {
expect(hasInstalledNodeModules(repoDir)).toBe(true);
expect(execFileSyncMock).not.toHaveBeenCalled();
});
it('disables corepack project specs only for lockfile-selected pnpm workspaces under a conflicting ancestor', () => {
const parentDir = path.join(tempRoot, 'parent');
fs.mkdirSync(parentDir, { recursive: true });
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'child');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({ name: 'child', scripts: { typecheck: 'tsc --noEmit' } }),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
'lockfileVersion: 9.0\n',
);
expect(
buildWorkspaceCommandEnvironment(repoDir, 'pnpm', { PATH: '/tmp/bin' }),
).toMatchObject({
PATH: '/tmp/bin',
COREPACK_ENABLE_PROJECT_SPEC: '0',
});
});
it('keeps an explicitly pinned pnpm workspace packageManager under a bun ancestor', () => {
const parentDir = path.join(tempRoot, 'parent');
fs.mkdirSync(parentDir, { recursive: true });
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'child');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({
name: 'child',
packageManager: 'pnpm@10.11.0',
scripts: { typecheck: 'tsc --noEmit' },
}),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
'lockfileVersion: 9.0\n',
);
expect(
buildWorkspaceCommandEnvironment(repoDir, 'pnpm', { PATH: '/tmp/bin' }),
).toEqual({ PATH: '/tmp/bin' });
});
});

View File

@@ -18,6 +18,7 @@ export interface WorkspaceDependencyInstallResult {
commandText?: string;
}
const COREPACK_PROJECT_SPEC_DISABLED = '0';
const INSTALL_STATE_FILENAME = '.ejclaw-install-state.json';
const NODE_MODULES_NOISE_ENTRIES = new Set([
'.cache',
@@ -117,6 +118,52 @@ function buildCorepackCommand(
};
}
function findNearestAncestorPackageManager(
repoDir: string,
): WorkspacePackageManager | null {
let currentDir = path.dirname(path.resolve(repoDir));
while (true) {
const packageManager = detectPackageManagerFromField(
readPackageJsonMetadata(currentDir)?.packageManager,
);
if (packageManager) {
return packageManager;
}
const parentDir = path.dirname(currentDir);
if (parentDir === currentDir) {
return null;
}
currentDir = parentDir;
}
}
export function buildWorkspaceCommandEnvironment(
repoDir: string,
packageManager: WorkspacePackageManager,
baseEnv: NodeJS.ProcessEnv = process.env,
): NodeJS.ProcessEnv {
if (packageManager !== 'pnpm') {
return baseEnv;
}
const localPackageManager = readPackageJsonMetadata(repoDir)?.packageManager;
if (typeof localPackageManager === 'string' && localPackageManager.trim()) {
return baseEnv;
}
const ancestorPackageManager = findNearestAncestorPackageManager(repoDir);
if (!ancestorPackageManager || ancestorPackageManager === packageManager) {
return baseEnv;
}
return {
...baseEnv,
COREPACK_ENABLE_PROJECT_SPEC: COREPACK_PROJECT_SPEC_DISABLED,
};
}
function computeInstallFingerprint(repoDir: string): string | null {
const packageJsonPath = path.join(repoDir, 'package.json');
if (!fs.existsSync(packageJsonPath)) {
@@ -359,6 +406,7 @@ export function ensureWorkspaceDependenciesInstalled(
try {
execFileSync(command.file, command.args, {
cwd: repoDir,
env: buildWorkspaceCommandEnvironment(repoDir, command.packageManager),
encoding: 'utf-8',
stdio: ['ignore', 'pipe', 'pipe'],
});
@@ -420,6 +468,7 @@ export function detectPnpmStorePath(workspaceDir: string): string | null {
try {
const storePath = execFileSync(file, args, {
cwd: workspaceDir,
env: buildWorkspaceCommandEnvironment(workspaceDir, 'pnpm'),
encoding: 'utf-8',
stdio: ['ignore', 'pipe', 'pipe'],
timeout: 5000,