claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add paired intent checkpoints

Browse Source
This commit is contained in:
Eyejoker
2026-03-29 03:34:55 +09:00
parent 2d8f8151dd
commit d460933229
10 changed files with 897 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
src/db.test.ts
View File

@@ -7,6 +7,7 @@ import { describe, it, expect, beforeEach } from 'vitest';
import {
_initTestDatabase,
_initTestDatabaseFromFile,
applyPairedEvent,
claimServiceHandoff,
completeServiceHandoffAndAdvanceTargetCursor,
createPairedApproval,
@@ -30,6 +31,7 @@ import {
getMessagesSince,
getNewMessages,
getPairedExecutionById,
getPairedEventByDedupeKey,
getPairedProject,
getPairedTaskById,
getPairedWorkspace,
@@ -39,6 +41,7 @@ import {
getTaskById,
listPairedApprovalsForTask,
listPairedArtifactsForTask,
listPairedEventsForTask,
listPairedWorkspacesForTask,
markWorkItemDelivered,
markWorkItemDeliveryRetry,
@@ -638,6 +641,181 @@ describe('paired task state', () => {
expect(listPairedArtifactsForTask('paired-task-1')[0]?.id).toBe(artifactId);
});
it('dedupes paired events and keeps request_review replay-safe', () => {
createPairedTask({
id: 'paired-task-events-1',
chat_jid: 'dc:paired',
group_folder: 'paired-room',
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: null,
status: 'active',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
const first = applyPairedEvent({
event: {
task_id: 'paired-task-events-1',
event_type: 'request_review',
actor_role: 'owner',
source_service_id: 'codex-main',
source_fingerprint: 'fingerprint-v1',
dedupe_key: 'msg-review-1',
payload_json: '{"request":"review"}',
created_at: '2026-03-29T00:01:00.000Z',
},
onApply: () => {
updatePairedTask('paired-task-events-1', {
status: 'review_pending',
review_requested_at: '2026-03-29T00:01:00.000Z',
updated_at: '2026-03-29T00:01:00.000Z',
});
},
});
const second = applyPairedEvent({
event: {
task_id: 'paired-task-events-1',
event_type: 'request_review',
actor_role: 'owner',
source_service_id: 'codex-main',
source_fingerprint: 'fingerprint-v2',
dedupe_key: 'msg-review-1',
payload_json: '{"request":"review-again"}',
created_at: '2026-03-29T00:02:00.000Z',
},
onApply: () => {
updatePairedTask('paired-task-events-1', {
status: 'review_ready',
review_requested_at: '2026-03-29T00:02:00.000Z',
updated_at: '2026-03-29T00:02:00.000Z',
});
},
});
expect(first.applied).toBe(true);
expect(second.applied).toBe(false);
expect(listPairedEventsForTask('paired-task-events-1')).toHaveLength(1);
expect(
getPairedEventByDedupeKey({
taskId: 'paired-task-events-1',
eventType: 'request_review',
dedupeKey: 'msg-review-1',
})?.source_fingerprint,
).toBe('fingerprint-v1');
expect(getPairedTaskById('paired-task-events-1')?.status).toBe(
'review_pending',
);
expect(getPairedTaskById('paired-task-events-1')?.review_requested_at).toBe(
'2026-03-29T00:01:00.000Z',
);
});
it('rolls back paired event insert when the coupled state update fails', () => {
createPairedTask({
id: 'paired-task-events-2',
chat_jid: 'dc:paired',
group_folder: 'paired-room',
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: null,
status: 'active',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
expect(() =>
applyPairedEvent({
event: {
task_id: 'paired-task-events-2',
event_type: 'set_risk',
actor_role: 'owner',
source_service_id: 'codex-main',
source_fingerprint: 'fingerprint-v1',
dedupe_key: 'msg-risk-1',
payload_json: '{"riskLevel":"high"}',
created_at: '2026-03-29T00:01:00.000Z',
},
onApply: () => {
updatePairedTask('paired-task-events-2', {
risk_level: 'high',
updated_at: '2026-03-29T00:01:00.000Z',
});
throw new Error('boom');
},
}),
).toThrow('boom');
expect(listPairedEventsForTask('paired-task-events-2')).toHaveLength(0);
expect(getPairedTaskById('paired-task-events-2')?.risk_level).toBe('low');
});
it('keeps paired event audit history after reopening the database', () => {
const tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-db-events-'));
const dbPath = path.join(tempRoot, 'messages.db');
_initTestDatabaseFromFile(dbPath);
createPairedTask({
id: 'paired-task-events-3',
chat_jid: 'dc:paired',
group_folder: 'paired-room',
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: null,
status: 'active',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
applyPairedEvent({
event: {
task_id: 'paired-task-events-3',
event_type: 'request_review',
actor_role: 'owner',
source_service_id: 'codex-main',
source_fingerprint: 'fingerprint-v1',
dedupe_key: 'msg-review-3',
payload_json: '{"request":"review"}',
created_at: '2026-03-29T00:01:00.000Z',
},
onApply: () => {
updatePairedTask('paired-task-events-3', {
status: 'review_pending',
review_requested_at: '2026-03-29T00:01:00.000Z',
updated_at: '2026-03-29T00:01:00.000Z',
});
},
});
_initTestDatabaseFromFile(dbPath);
expect(listPairedEventsForTask('paired-task-events-3')).toHaveLength(1);
expect(listPairedEventsForTask('paired-task-events-3')[0]?.dedupe_key).toBe(
'msg-review-3',
);
expect(
listPairedEventsForTask('paired-task-events-3')[0]?.source_fingerprint,
).toBe('fingerprint-v1');
fs.rmSync(tempRoot, { recursive: true, force: true });
});
it('backfills governance plan_status by legacy task status during migration', () => {
const tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-db-migration-'));
const dbPath = path.join(tempRoot, 'messages.db');

129
src/db.ts
View File

@@ -27,6 +27,7 @@ import {
AgentType,
PairedApproval,
PairedArtifact,
PairedEvent,
PairedExecution,
PairedProject,
PairedTask,
@@ -356,6 +357,31 @@ function createSchema(database: Database.Database): void {
);
CREATE INDEX IF NOT EXISTS idx_paired_artifacts_task
ON paired_artifacts(task_id, created_at);
CREATE TABLE IF NOT EXISTS paired_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT NOT NULL,
event_type TEXT NOT NULL,
actor_role TEXT NOT NULL,
source_service_id TEXT NOT NULL,
source_fingerprint TEXT,
dedupe_key TEXT NOT NULL,
payload_json TEXT,
created_at TEXT NOT NULL,
CHECK (
event_type IN (
'set_risk',
'submit_plan',
'approve_plan',
'request_plan_changes',
'request_review'
)
),
CHECK (actor_role IN ('owner', 'reviewer', 'system'))
);
CREATE INDEX IF NOT EXISTS idx_paired_events_task
ON paired_events(task_id, created_at, id);
CREATE UNIQUE INDEX IF NOT EXISTS idx_paired_events_dedupe
ON paired_events(task_id, event_type, dedupe_key);
CREATE TABLE IF NOT EXISTS channel_owner (
chat_jid TEXT PRIMARY KEY,
owner_service_id TEXT NOT NULL,
@@ -2325,6 +2351,109 @@ export function listPairedArtifactsForTask(taskId: string): PairedArtifact[] {
.all(taskId) as PairedArtifact[];
}
export function getPairedEventById(id: number): PairedEvent | undefined {
return db.prepare('SELECT * FROM paired_events WHERE id = ?').get(id) as
| PairedEvent
| undefined;
}
export function getPairedEventByDedupeKey(args: {
taskId: string;
eventType: PairedEvent['event_type'];
dedupeKey: string;
}): PairedEvent | undefined {
return db
.prepare(
`
SELECT *
FROM paired_events
WHERE task_id = ?
AND event_type = ?
AND dedupe_key = ?
`,
)
.get(args.taskId, args.eventType, args.dedupeKey) as
| PairedEvent
| undefined;
}
export function listPairedEventsForTask(taskId: string): PairedEvent[] {
return db
.prepare(
'SELECT * FROM paired_events WHERE task_id = ? ORDER BY created_at, id',
)
.all(taskId) as PairedEvent[];
}
export function applyPairedEvent<T>(args: {
event: Omit<PairedEvent, 'id'>;
onApply?: () => T;
}): {
applied: boolean;
event: PairedEvent;
result: T | null;
} {
return db.transaction(() => {
const insertResult = db
.prepare(
`
INSERT OR IGNORE INTO paired_events (
task_id,
event_type,
actor_role,
source_service_id,
source_fingerprint,
dedupe_key,
payload_json,
created_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`,
)
.run(
args.event.task_id,
args.event.event_type,
args.event.actor_role,
args.event.source_service_id,
args.event.source_fingerprint,
args.event.dedupe_key,
args.event.payload_json,
args.event.created_at,
);
if (insertResult.changes === 0) {
const existing = getPairedEventByDedupeKey({
taskId: args.event.task_id,
eventType: args.event.event_type,
dedupeKey: args.event.dedupe_key,
});
if (!existing) {
throw new Error(
`Paired event dedupe lookup failed for ${args.event.task_id}:${args.event.event_type}:${args.event.dedupe_key}`,
);
}
return {
applied: false,
event: existing,
result: null as T | null,
};
}
const inserted = getPairedEventById(Number(insertResult.lastInsertRowid));
if (!inserted) {
throw new Error(
`Paired event insert lookup failed for row ${insertResult.lastInsertRowid}`,
);
}
return {
applied: true,
event: inserted,
result: args.onApply ? args.onApply() : (null as T | null),
};
})();
}
/**
* Get the most recent bot message (is_bot_message=1) in a chat, regardless of which bot sent it.
* Used for duplicate detection in pair rooms.

15
src/message-runtime.ts
View File

@@ -572,7 +572,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
// Session commands are explicit control events, so owner-only and
// reviewer-only commands resolve against the target role rather than
// whichever service happened to pick up the slash command first.
markReviewReady: async () => {
markReviewReady: async (dedupeKey) => {
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(
lease,
@@ -582,10 +582,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
group,
chatJid,
roomRoleContext,
dedupeKey,
});
return formatRoomReviewReadyMessage(result);
},
setTaskRiskLevel: async (riskLevel) => {
setTaskRiskLevel: async (riskLevel, dedupeKey) => {
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(
lease,
@@ -596,9 +597,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
chatJid,
roomRoleContext,
riskLevel,
dedupeKey,
});
},
recordPlan: async (plan) => {
recordPlan: async (plan, dedupeKey) => {
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(
lease,
@@ -608,10 +610,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
group,
chatJid,
roomRoleContext,
dedupeKey,
...plan,
});
},
approvePlan: async () => {
approvePlan: async (dedupeKey) => {
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(
lease,
@@ -621,9 +624,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
group,
chatJid,
roomRoleContext,
dedupeKey,
});
},
requestPlanChanges: async (note) => {
requestPlanChanges: async (note, dedupeKey) => {
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(
lease,
@@ -634,6 +638,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
chatJid,
roomRoleContext,
note,
dedupeKey,
});
},
},

232
src/paired-execution-context.test.ts
View File

@@ -1,6 +1,7 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('./db.js', () => ({
applyPairedEvent: vi.fn(),
createPairedArtifact: vi.fn(),
createPairedExecution: vi.fn(),
createPairedTask: vi.fn(),
@@ -20,6 +21,7 @@ vi.mock('./paired-workspace-manager.js', () => ({
markPairedTaskReviewReady: vi.fn(),
prepareReviewerWorkspaceForExecution: vi.fn(),
provisionOwnerWorkspaceForPairedTask: vi.fn(),
resolvePairedTaskSourceFingerprint: vi.fn(),
}));
vi.mock('./logger.js', () => ({
@@ -69,9 +71,101 @@ const reviewerContext: RoomRoleContext = {
failoverOwner: false,
};
async function importExecutionContextForService(serviceId: string) {
vi.resetModules();
const dbModule = {
applyPairedEvent: vi.fn(),
createPairedArtifact: vi.fn(),
createPairedExecution: vi.fn(),
createPairedTask: vi.fn(),
getLatestOpenPairedTaskForChat: vi.fn(),
getPairedExecutionById: vi.fn(),
getPairedTaskById: vi.fn(),
getPairedWorkspace: vi.fn(),
listPairedArtifactsForTask: vi.fn(),
updatePairedExecution: vi.fn(),
updatePairedTask: vi.fn(),
upsertPairedProject: vi.fn(),
};
dbModule.applyPairedEvent.mockImplementation(({ event, onApply }) => ({
applied: true,
event: { id: 1, ...event },
result: onApply ? onApply() : null,
}));
dbModule.getLatestOpenPairedTaskForChat.mockReturnValue(undefined);
dbModule.getPairedExecutionById.mockReturnValue(undefined);
dbModule.getPairedTaskById.mockReturnValue(undefined);
dbModule.getPairedWorkspace.mockReturnValue(undefined);
dbModule.listPairedArtifactsForTask.mockReturnValue([]);
const pairedWorkspaceManagerModule = {
PLAN_REVIEW_REQUIRED_BLOCK_MESSAGE:
'Plan review is required before formal review for this high-risk task.',
markPairedTaskReviewReady: vi.fn(),
prepareReviewerWorkspaceForExecution: vi.fn(),
provisionOwnerWorkspaceForPairedTask: vi.fn(),
resolvePairedTaskSourceFingerprint: vi.fn(),
};
pairedWorkspaceManagerModule.provisionOwnerWorkspaceForPairedTask.mockReturnValue(
{
id: 'task-1:owner',
task_id: 'task-1',
role: 'owner',
workspace_dir: '/tmp/paired/task-1/owner',
snapshot_source_dir: null,
snapshot_source_fingerprint: null,
status: 'ready',
snapshot_refreshed_at: null,
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
},
);
pairedWorkspaceManagerModule.prepareReviewerWorkspaceForExecution.mockReturnValue(
{
workspace: null,
autoRefreshed: false,
},
);
pairedWorkspaceManagerModule.resolvePairedTaskSourceFingerprint.mockReturnValue(
'fingerprint-1',
);
vi.doMock('./db.js', () => dbModule);
vi.doMock('./paired-workspace-manager.js', () => pairedWorkspaceManagerModule);
vi.doMock('./logger.js', () => ({
logger: {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
},
}));
vi.doMock('./config.js', async () => {
const actual =
await vi.importActual<typeof import('./config.js')>('./config.js');
return {
...actual,
SERVICE_ID: serviceId,
};
});
const module = await import('./paired-execution-context.js');
return {
dbModule,
pairedWorkspaceManagerModule,
markRoomReviewReady: module.markRoomReviewReady,
};
}
describe('paired execution context', () => {
beforeEach(() => {
vi.resetAllMocks();
vi.mocked(db.applyPairedEvent).mockImplementation(({ event, onApply }) => ({
applied: true,
event: { id: 1, ...event },
result: onApply ? onApply() : null,
}));
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue(undefined);
vi.mocked(db.getPairedExecutionById).mockReturnValue(undefined);
vi.mocked(db.getPairedTaskById).mockReturnValue(undefined);
@@ -97,6 +191,9 @@ describe('paired execution context', () => {
workspace: null,
autoRefreshed: false,
});
vi.mocked(
pairedWorkspaceManager.resolvePairedTaskSourceFingerprint,
).mockReturnValue('fingerprint-1');
});
it('creates an owner execution with a worktree override', () => {
@@ -280,8 +377,14 @@ describe('paired execution context', () => {
expect(result?.envOverrides.EJCLAW_WORK_DIR).toBeUndefined();
});
it('marks the active room task review-ready through the workspace manager', () => {
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
it('returns ready for /review on the owner service', async () => {
const {
dbModule,
pairedWorkspaceManagerModule,
markRoomReviewReady: isolatedMarkRoomReviewReady,
} = await importExecutionContextForService('codex-main');
dbModule.getLatestOpenPairedTaskForChat.mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
@@ -297,7 +400,7 @@ describe('paired execution context', () => {
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
});
vi.mocked(db.getPairedTaskById).mockReturnValue({
dbModule.getPairedTaskById.mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
@@ -313,43 +416,44 @@ describe('paired execution context', () => {
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
});
vi.mocked(pairedWorkspaceManager.markPairedTaskReviewReady).mockReturnValue(
{
ownerWorkspace: {
id: 'task-1:owner',
task_id: 'task-1',
role: 'owner',
workspace_dir: '/tmp/paired/task-1/owner',
snapshot_source_dir: null,
snapshot_source_fingerprint: null,
status: 'ready',
snapshot_refreshed_at: null,
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
},
reviewerWorkspace: {
id: 'task-1:reviewer',
task_id: 'task-1',
role: 'reviewer',
workspace_dir: '/tmp/paired/task-1/reviewer',
snapshot_source_dir: '/tmp/paired/task-1/owner',
snapshot_source_fingerprint: 'fingerprint-1',
status: 'ready',
snapshot_refreshed_at: '2026-03-28T00:00:00.000Z',
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
},
pairedWorkspaceManagerModule.markPairedTaskReviewReady.mockReturnValue({
ownerWorkspace: {
id: 'task-1:owner',
task_id: 'task-1',
role: 'owner',
workspace_dir: '/tmp/paired/task-1/owner',
snapshot_source_dir: null,
snapshot_source_fingerprint: null,
status: 'ready',
snapshot_refreshed_at: null,
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
},
);
reviewerWorkspace: {
id: 'task-1:reviewer',
task_id: 'task-1',
role: 'reviewer',
workspace_dir: '/tmp/paired/task-1/reviewer',
snapshot_source_dir: '/tmp/paired/task-1/owner',
snapshot_source_fingerprint: 'fingerprint-1',
status: 'ready',
snapshot_refreshed_at: '2026-03-28T00:00:00.000Z',
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
},
});
const result = markRoomReviewReady({
const result = isolatedMarkRoomReviewReady({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
});
expect(
pairedWorkspaceManager.markPairedTaskReviewReady,
pairedWorkspaceManagerModule.provisionOwnerWorkspaceForPairedTask,
).toHaveBeenCalledWith('task-1');
expect(
pairedWorkspaceManagerModule.markPairedTaskReviewReady,
).toHaveBeenCalledWith('task-1');
expect(result?.status).toBe('ready');
if (!result || result.status !== 'ready') {
@@ -358,6 +462,70 @@ describe('paired execution context', () => {
expect(result.task.status).toBe('review_ready');
});
it('returns pending for /review on the reviewer service before the owner workspace is visible', async () => {
const {
dbModule,
pairedWorkspaceManagerModule,
markRoomReviewReady: isolatedMarkRoomReviewReady,
} = await importExecutionContextForService('codex-review');
dbModule.getLatestOpenPairedTaskForChat.mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: null,
status: 'draft',
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
});
dbModule.getPairedTaskById.mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-28T00:00:00.000Z',
status: 'review_pending',
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
});
pairedWorkspaceManagerModule.markPairedTaskReviewReady.mockReturnValue(null);
const result = isolatedMarkRoomReviewReady({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
});
expect(
pairedWorkspaceManagerModule.provisionOwnerWorkspaceForPairedTask,
).not.toHaveBeenCalled();
expect(dbModule.getPairedWorkspace).toHaveBeenCalledWith('task-1', 'owner');
expect(
pairedWorkspaceManagerModule.markPairedTaskReviewReady,
).toHaveBeenCalledWith('task-1');
expect(result).toEqual({
status: 'pending',
task: expect.objectContaining({
id: 'task-1',
status: 'review_pending',
}),
pendingReason: 'owner-workspace-not-ready',
});
});
it('keeps review_pending and returns a pending result when owner workspace is not ready', () => {
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
id: 'task-1',

269
src/paired-execution-context.ts
View File

@@ -3,12 +3,14 @@ import crypto from 'crypto';
import { SERVICE_ID, normalizeServiceId } from './config.js';
import {
applyPairedEvent,
createPairedArtifact,
createPairedExecution,
createPairedTask,
getLatestOpenPairedTaskForChat,
getPairedExecutionById,
getPairedTaskById,
getPairedWorkspace,
listPairedArtifactsForTask,
updatePairedExecution,
updatePairedTask,
@@ -20,9 +22,11 @@ import {
markPairedTaskReviewReady,
prepareReviewerWorkspaceForExecution,
provisionOwnerWorkspaceForPairedTask,
resolvePairedTaskSourceFingerprint,
} from './paired-workspace-manager.js';
import type {
PairedArtifactType,
PairedEventType,
PairedExecution,
PairedTask,
PairedWorkspace,
@@ -249,6 +253,51 @@ const PLAN_REQUIRES_HIGH_RISK_MESSAGE =
const PLAN_INCOMPLETE_MESSAGE =
'Plan artifacts are incomplete. Record /plan <plan brief> || <acceptance criteria> || <risk summary> before approval.';
function normalizeIntentDedupeKey(dedupeKey?: string): string {
return dedupeKey?.trim() || crypto.randomUUID();
}
function serializeIntentPayload(
payload?: Record<string, unknown>,
): string | null {
if (!payload || Object.keys(payload).length === 0) {
return null;
}
return JSON.stringify(payload);
}
function applyRoomIntent<T>(args: {
task: PairedTask;
roomRoleContext: RoomRoleContext;
eventType: PairedEventType;
dedupeKey?: string;
payload?: Record<string, unknown>;
onApply?: () => T;
}): {
applied: boolean;
result: T | null;
} {
const createdAt = new Date().toISOString();
const sourceFingerprint = resolvePairedTaskSourceFingerprint(args.task.id);
const result = applyPairedEvent({
event: {
task_id: args.task.id,
event_type: args.eventType,
actor_role: args.roomRoleContext.role,
source_service_id: args.roomRoleContext.serviceId,
source_fingerprint: sourceFingerprint,
dedupe_key: normalizeIntentDedupeKey(args.dedupeKey),
payload_json: serializeIntentPayload(args.payload),
created_at: createdAt,
},
onApply: args.onApply,
});
return {
applied: result.applied,
result: result.result,
};
}
function isPlanReviewRequired(task: PairedTask): boolean {
return task.risk_level === 'high' && task.plan_status !== 'approved';
}
@@ -289,6 +338,16 @@ function createTaskArtifact(args: {
});
}
function getPersistedReviewReadyState(taskId: string): {
ownerWorkspace: PairedWorkspace | null;
reviewerWorkspace: PairedWorkspace | null;
} {
return {
ownerWorkspace: getPairedWorkspace(taskId, 'owner') ?? null,
reviewerWorkspace: getPairedWorkspace(taskId, 'reviewer') ?? null,
};
}
function formatPlanReviewRequiredMessage(task: PairedTask): string {
return [
PLAN_REVIEW_REQUIRED_BLOCK_MESSAGE,
@@ -314,8 +373,9 @@ export function markRoomReviewReady(args: {
group: RegisteredGroup;
chatJid: string;
roomRoleContext?: RoomRoleContext;
dedupeKey?: string;
}): MarkRoomReviewReadyResult {
const { group, chatJid, roomRoleContext } = args;
const { group, chatJid, roomRoleContext, dedupeKey } = args;
if (!roomRoleContext || !group.workDir) {
return null;
}
@@ -335,12 +395,13 @@ export function markRoomReviewReady(args: {
const isOwnerService =
normalizeServiceId(SERVICE_ID) === task.owner_service_id;
if (isOwnerService) {
provisionOwnerWorkspaceForPairedTask(task.id);
}
const ownerWorkspace =
(isOwnerService
? provisionOwnerWorkspaceForPairedTask(task.id)
: getPairedWorkspace(task.id, 'owner')) ?? null;
const reviewReady = markPairedTaskReviewReady(task.id);
if (!reviewReady) {
if (!ownerWorkspace) {
markPairedTaskReviewReady(task.id);
return {
status: 'pending',
task: getPairedTaskById(task.id) ?? task,
@@ -348,11 +409,40 @@ export function markRoomReviewReady(args: {
};
}
const { ownerWorkspace, reviewerWorkspace } = reviewReady;
const reviewIntent = applyRoomIntent({
task,
roomRoleContext,
eventType: 'request_review',
dedupeKey,
onApply: () => markPairedTaskReviewReady(task.id),
});
const latestTask = getPairedTaskById(task.id) ?? task;
const { ownerWorkspace: persistedOwnerWorkspace, reviewerWorkspace } =
getPersistedReviewReadyState(
task.id,
);
if (reviewIntent.result) {
return {
status: 'ready',
task: latestTask,
ownerWorkspace: reviewIntent.result.ownerWorkspace,
reviewerWorkspace: reviewIntent.result.reviewerWorkspace,
};
}
if (!ownerWorkspace || !reviewerWorkspace) {
return {
status: 'pending',
task: latestTask,
pendingReason: 'owner-workspace-not-ready',
};
}
return {
status: 'ready',
task: getPairedTaskById(task.id) ?? task,
ownerWorkspace,
task: latestTask,
ownerWorkspace: persistedOwnerWorkspace ?? ownerWorkspace,
reviewerWorkspace,
};
}
@@ -389,8 +479,9 @@ export function setRoomTaskRiskLevel(args: {
chatJid: string;
roomRoleContext?: RoomRoleContext;
riskLevel: 'low' | 'high';
dedupeKey?: string;
}): string | null {
const { group, chatJid, roomRoleContext, riskLevel } = args;
const { group, chatJid, roomRoleContext, riskLevel, dedupeKey } = args;
if (!roomRoleContext || !group.workDir) {
return null;
}
@@ -403,23 +494,33 @@ export function setRoomTaskRiskLevel(args: {
return null;
}
const now = new Date().toISOString();
if (riskLevel === 'high') {
updatePairedTask(task.id, {
risk_level: 'high',
plan_status: hasCompletePlanArtifacts(task.id)
? 'pending'
: 'not_requested',
status: 'plan_review_pending',
updated_at: now,
});
} else {
updatePairedTask(task.id, {
risk_level: 'low',
status: task.status === 'plan_review_pending' ? 'active' : task.status,
updated_at: now,
});
}
applyRoomIntent({
task,
roomRoleContext,
eventType: 'set_risk',
dedupeKey,
payload: { riskLevel },
onApply: () => {
const now = new Date().toISOString();
if (riskLevel === 'high') {
updatePairedTask(task.id, {
risk_level: 'high',
plan_status: hasCompletePlanArtifacts(task.id)
? 'pending'
: 'not_requested',
status: 'plan_review_pending',
updated_at: now,
});
return;
}
updatePairedTask(task.id, {
risk_level: 'low',
status: task.status === 'plan_review_pending' ? 'active' : task.status,
updated_at: now,
});
},
});
const latestTask = getPairedTaskById(task.id) ?? task;
return [
@@ -437,6 +538,7 @@ export function recordRoomPlan(args: {
planBrief: string;
acceptanceCriteria: string;
riskSummary: string;
dedupeKey?: string;
}): string | null {
const {
group,
@@ -445,6 +547,7 @@ export function recordRoomPlan(args: {
planBrief,
acceptanceCriteria,
riskSummary,
dedupeKey,
} = args;
if (!roomRoleContext || !group.workDir) {
return null;
@@ -461,29 +564,42 @@ export function recordRoomPlan(args: {
return PLAN_REQUIRES_HIGH_RISK_MESSAGE;
}
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'plan_brief',
content: planBrief,
});
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'acceptance_criteria',
content: acceptanceCriteria,
});
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'risk_summary',
content: riskSummary,
});
applyRoomIntent({
task,
roomRoleContext,
eventType: 'submit_plan',
dedupeKey,
payload: {
planBrief,
acceptanceCriteria,
riskSummary,
},
onApply: () => {
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'plan_brief',
content: planBrief,
});
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'acceptance_criteria',
content: acceptanceCriteria,
});
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'risk_summary',
content: riskSummary,
});
updatePairedTask(task.id, {
plan_status: 'pending',
status: 'plan_review_pending',
updated_at: new Date().toISOString(),
updatePairedTask(task.id, {
plan_status: 'pending',
status: 'plan_review_pending',
updated_at: new Date().toISOString(),
});
},
});
const latestTask = getPairedTaskById(task.id) ?? task;
@@ -499,8 +615,9 @@ export function approveRoomPlan(args: {
group: RegisteredGroup;
chatJid: string;
roomRoleContext?: RoomRoleContext;
dedupeKey?: string;
}): string | null {
const { group, chatJid, roomRoleContext } = args;
const { group, chatJid, roomRoleContext, dedupeKey } = args;
if (!roomRoleContext || !group.workDir) {
return null;
}
@@ -519,10 +636,18 @@ export function approveRoomPlan(args: {
return PLAN_INCOMPLETE_MESSAGE;
}
updatePairedTask(task.id, {
plan_status: 'approved',
status: 'active',
updated_at: new Date().toISOString(),
applyRoomIntent({
task,
roomRoleContext,
eventType: 'approve_plan',
dedupeKey,
onApply: () => {
updatePairedTask(task.id, {
plan_status: 'approved',
status: 'active',
updated_at: new Date().toISOString(),
});
},
});
const latestTask = getPairedTaskById(task.id) ?? task;
@@ -538,8 +663,9 @@ export function requestRoomPlanChanges(args: {
chatJid: string;
roomRoleContext?: RoomRoleContext;
note?: string;
dedupeKey?: string;
}): string | null {
const { group, chatJid, roomRoleContext, note } = args;
const { group, chatJid, roomRoleContext, note, dedupeKey } = args;
if (!roomRoleContext || !group.workDir) {
return null;
}
@@ -555,19 +681,28 @@ export function requestRoomPlanChanges(args: {
return PLAN_REQUIRES_HIGH_RISK_MESSAGE;
}
if (note?.trim()) {
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'comment',
content: note.trim(),
});
}
applyRoomIntent({
task,
roomRoleContext,
eventType: 'request_plan_changes',
dedupeKey,
payload: note?.trim() ? { note: note.trim() } : undefined,
onApply: () => {
if (note?.trim()) {
createTaskArtifact({
taskId: task.id,
serviceId: roomRoleContext.serviceId,
artifactType: 'comment',
content: note.trim(),
});
}
updatePairedTask(task.id, {
plan_status: 'changes_requested',
status: 'plan_review_pending',
updated_at: new Date().toISOString(),
updatePairedTask(task.id, {
plan_status: 'changes_requested',
status: 'plan_review_pending',
updated_at: new Date().toISOString(),
});
},
});
const latestTask = getPairedTaskById(task.id) ?? task;

113
src/paired-review-command.test.ts
View File

@@ -66,6 +66,55 @@ describe('paired /review command path', () => {
fs.rmSync(tempRoot, { recursive: true, force: true });
});
it('records the first owner /review event with the provisioned owner workspace fingerprint', async () => {
const canonicalDir = path.join(tempRoot, 'canonical');
fs.mkdirSync(canonicalDir, { recursive: true });
runGit(['init'], canonicalDir);
runGit(['config', 'user.email', 'test@example.com'], canonicalDir);
runGit(['config', 'user.name', 'EJClaw Test'], canonicalDir);
fs.writeFileSync(path.join(canonicalDir, 'README.md'), 'original\n');
runGit(['add', 'README.md'], canonicalDir);
runGit(['commit', '-m', 'initial'], canonicalDir);
const group: RegisteredGroup = {
name: 'Paired Room',
folder: 'paired-room',
trigger: '@codex',
added_at: '2026-03-28T00:00:00.000Z',
agentType: 'codex',
workDir: canonicalDir,
};
process.env.EJCLAW_DATA_DIR = path.join(tempRoot, 'data-owner');
process.env.SERVICE_ID = 'codex-main';
vi.resetModules();
const { db, executionContext } = await loadModules();
db.initDatabase();
const result = executionContext.markRoomReviewReady({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
dedupeKey: 'msg-review-owner-1',
});
const task = db.getLatestOpenPairedTaskForChat('dc:test');
const events = db.listPairedEventsForTask(task!.id);
expect(result?.status).toBe('ready');
expect(events).toHaveLength(1);
expect(events[0]?.event_type).toBe('request_review');
expect(events[0]?.dedupe_key).toBe('msg-review-owner-1');
expect(events[0]?.source_fingerprint).toBeTruthy();
expect(events[0]?.source_fingerprint).not.toBe(task?.source_ref);
if (!result || result.status !== 'ready') {
throw new Error('expected owner /review to prepare a ready snapshot');
}
expect(events[0]?.source_fingerprint).toBe(
result.reviewerWorkspace.snapshot_source_fingerprint,
);
});
it('reuses the db owner workspace when reviewer service handles /review', async () => {
const canonicalDir = path.join(tempRoot, 'canonical');
fs.mkdirSync(canonicalDir, { recursive: true });
@@ -197,9 +246,73 @@ describe('paired /review command path', () => {
});
expect(task?.status).toBe('review_pending');
expect(task?.review_requested_at).toBeTruthy();
expect(db.listPairedEventsForTask(task!.id)).toEqual([]);
expect(fs.existsSync(reviewerLocalOwnerDir)).toBe(false);
});
it('creates the first checkpointed request_review event only after owner workspace exists', async () => {
const canonicalDir = path.join(tempRoot, 'canonical');
fs.mkdirSync(canonicalDir, { recursive: true });
runGit(['init'], canonicalDir);
runGit(['config', 'user.email', 'test@example.com'], canonicalDir);
runGit(['config', 'user.name', 'EJClaw Test'], canonicalDir);
fs.writeFileSync(path.join(canonicalDir, 'README.md'), 'original\n');
runGit(['add', 'README.md'], canonicalDir);
runGit(['commit', '-m', 'initial'], canonicalDir);
const group: RegisteredGroup = {
name: 'Paired Room',
folder: 'paired-room',
trigger: '@codex',
added_at: '2026-03-28T00:00:00.000Z',
agentType: 'codex',
workDir: canonicalDir,
};
process.env.EJCLAW_DATA_DIR = path.join(tempRoot, 'data-review');
process.env.SERVICE_ID = 'codex-review';
vi.resetModules();
let { db, executionContext } = await loadModules();
db.initDatabase();
const firstResult = executionContext.markRoomReviewReady({
group,
chatJid: 'dc:test',
roomRoleContext: reviewerContext,
dedupeKey: 'msg-review-pending-1',
});
const task = db.getLatestOpenPairedTaskForChat('dc:test');
expect(firstResult?.status).toBe('pending');
expect(db.listPairedEventsForTask(task!.id)).toEqual([]);
process.env.EJCLAW_DATA_DIR = path.join(tempRoot, 'data-owner');
process.env.SERVICE_ID = 'codex-main';
vi.resetModules();
({ db, executionContext } = await loadModules());
db.initDatabase();
executionContext.preparePairedExecutionContext({
group,
chatJid: 'dc:test',
runId: 'run-owner',
roomRoleContext: ownerContext,
});
const secondResult = executionContext.markRoomReviewReady({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
dedupeKey: 'msg-review-owner-2',
});
const events = db.listPairedEventsForTask(task!.id);
expect(secondResult?.status).toBe('ready');
expect(events).toHaveLength(1);
expect(events[0]?.event_type).toBe('request_review');
expect(events[0]?.dedupe_key).toBe('msg-review-owner-2');
});
it('blocks /review when a high-risk task plan is not approved yet', async () => {
const canonicalDir = path.join(tempRoot, 'canonical');
fs.mkdirSync(canonicalDir, { recursive: true });

18
src/paired-workspace-manager.ts
View File

@@ -329,6 +329,24 @@ function makeWorkspaceRecord(args: {
};
}
export function resolvePairedTaskSourceFingerprint(taskId: string): string | null {
const { task } = getTaskAndProject(taskId);
const ownerWorkspace = getPairedWorkspace(taskId, 'owner');
if (!ownerWorkspace) {
return task.source_ref || null;
}
ensureGitRepository(ownerWorkspace.workspace_dir);
return buildReviewerSnapshotFingerprint({
sourceDir: ownerWorkspace.workspace_dir,
allowedTrackedFiles: listAllowedTrackedFiles(ownerWorkspace.workspace_dir),
deletedTrackedFiles: listDeletedTrackedFiles(ownerWorkspace.workspace_dir),
allowedUntrackedFiles: listAllowedUntrackedFiles(
ownerWorkspace.workspace_dir,
),
});
}
export function provisionOwnerWorkspaceForPairedTask(
taskId: string,
): PairedWorkspace {

17
src/session-commands.test.ts
View File

@@ -263,7 +263,7 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.markReviewReady).toHaveBeenCalledTimes(1);
expect(deps.markReviewReady).toHaveBeenCalledWith('msg-1');
expect(deps.runAgent).not.toHaveBeenCalled();
expect(deps.advanceCursor).toHaveBeenCalledWith('100');
expect(deps.sendMessage).toHaveBeenCalledWith(
@@ -288,7 +288,7 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.markReviewReady).toHaveBeenCalledTimes(1);
expect(deps.markReviewReady).toHaveBeenCalledWith('msg-1');
expect(deps.runAgent).not.toHaveBeenCalled();
expect(deps.advanceCursor).toHaveBeenCalledWith('100');
expect(deps.sendMessage).toHaveBeenCalledWith(
@@ -319,7 +319,7 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.markReviewReady).toHaveBeenCalledTimes(1);
expect(deps.markReviewReady).toHaveBeenCalledWith('msg-1');
expect(deps.sendMessage).toHaveBeenCalledWith(
[
'Review request recorded, but the owner workspace is not ready yet.',
@@ -346,7 +346,7 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.setTaskRiskLevel).toHaveBeenCalledWith('high');
expect(deps.setTaskRiskLevel).toHaveBeenCalledWith('high', 'msg-1');
expect(deps.runAgent).not.toHaveBeenCalled();
expect(deps.sendMessage).toHaveBeenCalledWith(
'Task risk updated.\n- Task: paired-task-1',
@@ -376,7 +376,7 @@ describe('handleSessionCommand', () => {
planBrief: 'brief',
acceptanceCriteria: 'acceptance criteria',
riskSummary: 'rollout risk',
});
}, 'msg-1');
expect(deps.sendMessage).toHaveBeenCalledWith(
'Plan recorded.\n- Task: paired-task-1',
);
@@ -418,7 +418,7 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.approvePlan).toHaveBeenCalledTimes(1);
expect(deps.approvePlan).toHaveBeenCalledWith('msg-1');
expect(deps.sendMessage).toHaveBeenCalledWith(
'Plan approved.\n- Task: paired-task-1',
);
@@ -441,7 +441,10 @@ describe('handleSessionCommand', () => {
});
expect(result).toEqual({ handled: true, success: true });
expect(deps.requestPlanChanges).toHaveBeenCalledWith('tighten scope');
expect(deps.requestPlanChanges).toHaveBeenCalledWith(
'tighten scope',
'msg-1',
);
expect(deps.sendMessage).toHaveBeenCalledWith(
'Plan changes requested.\n- Task: paired-task-1',
);

26
src/session-commands.ts
View File

@@ -94,15 +94,21 @@ export interface SessionCommandDeps {
isAdminSender: (msg: NewMessage) => boolean;
/** Whether the denied sender would normally be allowed to interact (for denial messages). */
canSenderInteract: (msg: NewMessage) => boolean;
markReviewReady: () => Promise<string | null>;
setTaskRiskLevel: (riskLevel: 'low' | 'high') => Promise<string | null>;
markReviewReady: (dedupeKey: string) => Promise<string | null>;
setTaskRiskLevel: (
riskLevel: 'low' | 'high',
dedupeKey: string,
) => Promise<string | null>;
recordPlan: (plan: {
planBrief: string;
acceptanceCriteria: string;
riskSummary: string;
}) => Promise<string | null>;
approvePlan: () => Promise<string | null>;
requestPlanChanges: (note?: string) => Promise<string | null>;
}, dedupeKey: string) => Promise<string | null>;
approvePlan: (dedupeKey: string) => Promise<string | null>;
requestPlanChanges: (
note: string | undefined,
dedupeKey: string,
) => Promise<string | null>;
}
function resultToText(result: string | object | null | undefined): string {
@@ -188,7 +194,7 @@ export async function handleSessionCommand(opts: {
}
if (command === '/review') {
const result = await deps.markReviewReady();
const result = await deps.markReviewReady(cmdMsg.id);
deps.advanceCursor(cmdMsg.timestamp);
await deps.sendMessage(
result ??
@@ -207,7 +213,7 @@ export async function handleSessionCommand(opts: {
deps.advanceCursor(cmdMsg.timestamp);
await deps.sendMessage(
riskLevel
? ((await deps.setTaskRiskLevel(riskLevel)) ??
? ((await deps.setTaskRiskLevel(riskLevel, cmdMsg.id)) ??
'Risk is unavailable for this room.')
: 'Usage: /risk <low|high>',
);
@@ -229,7 +235,7 @@ export async function handleSessionCommand(opts: {
planBrief: parts[0],
acceptanceCriteria: parts[1],
riskSummary: parts[2],
})) ?? 'Plan recording is unavailable for this room.',
}, cmdMsg.id)) ?? 'Plan recording is unavailable for this room.',
);
return { handled: true, success: true };
}
@@ -237,7 +243,7 @@ export async function handleSessionCommand(opts: {
if (command === '/approve-plan') {
deps.advanceCursor(cmdMsg.timestamp);
await deps.sendMessage(
(await deps.approvePlan()) ??
(await deps.approvePlan(cmdMsg.id)) ??
'Plan approval is unavailable for this room.',
);
return { handled: true, success: true };
@@ -249,7 +255,7 @@ export async function handleSessionCommand(opts: {
.trim();
deps.advanceCursor(cmdMsg.timestamp);
await deps.sendMessage(
(await deps.requestPlanChanges(note || undefined)) ??
(await deps.requestPlanChanges(note || undefined, cmdMsg.id)) ??
'Plan change requests are unavailable for this room.',
);
return { handled: true, success: true };

21
src/types.ts
View File

@@ -76,6 +76,15 @@ export type PairedArtifactType =
| 'acceptance_criteria'
| 'risk_summary';
export type PairedEventActorRole = PairedRoomRole | 'system';
export type PairedEventType =
| 'set_risk'
| 'submit_plan'
| 'approve_plan'
| 'request_plan_changes'
| 'request_review';
export interface RoomRoleContext {
serviceId: string;
role: PairedRoomRole;
@@ -158,6 +167,18 @@ export interface PairedArtifact {
created_at: string;
}
export interface PairedEvent {
id: number;
task_id: string;
event_type: PairedEventType;
actor_role: PairedEventActorRole;
source_service_id: string;
source_fingerprint: string | null;
dedupe_key: string;
payload_json: string | null;
created_at: string;
}
export type StructuredAgentOutput =
| {
visibility: 'public';