diff --git a/src/db.test.ts b/src/db.test.ts index 123eb88..1ccd256 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -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'); diff --git a/src/db.ts b/src/db.ts index 6704451..b995e97 100644 --- a/src/db.ts +++ b/src/db.ts @@ -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(args: { + event: Omit; + 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. diff --git a/src/message-runtime.ts b/src/message-runtime.ts index e4209bf..572902f 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -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, }); }, }, diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index 65e2d47..c2bc9f6 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -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('./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', diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index bd7b046..d8d4315 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -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 || || before approval.'; +function normalizeIntentDedupeKey(dedupeKey?: string): string { + return dedupeKey?.trim() || crypto.randomUUID(); +} + +function serializeIntentPayload( + payload?: Record, +): string | null { + if (!payload || Object.keys(payload).length === 0) { + return null; + } + return JSON.stringify(payload); +} + +function applyRoomIntent(args: { + task: PairedTask; + roomRoleContext: RoomRoleContext; + eventType: PairedEventType; + dedupeKey?: string; + payload?: Record; + 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; diff --git a/src/paired-review-command.test.ts b/src/paired-review-command.test.ts index 8645a3f..9bbfef6 100644 --- a/src/paired-review-command.test.ts +++ b/src/paired-review-command.test.ts @@ -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 }); diff --git a/src/paired-workspace-manager.ts b/src/paired-workspace-manager.ts index 110db50..9864f09 100644 --- a/src/paired-workspace-manager.ts +++ b/src/paired-workspace-manager.ts @@ -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 { diff --git a/src/session-commands.test.ts b/src/session-commands.test.ts index 806b2b8..cf52a61 100644 --- a/src/session-commands.test.ts +++ b/src/session-commands.test.ts @@ -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', ); diff --git a/src/session-commands.ts b/src/session-commands.ts index fd7919d..9ed501a 100644 --- a/src/session-commands.ts +++ b/src/session-commands.ts @@ -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; - setTaskRiskLevel: (riskLevel: 'low' | 'high') => Promise; + markReviewReady: (dedupeKey: string) => Promise; + setTaskRiskLevel: ( + riskLevel: 'low' | 'high', + dedupeKey: string, + ) => Promise; recordPlan: (plan: { planBrief: string; acceptanceCriteria: string; riskSummary: string; - }) => Promise; - approvePlan: () => Promise; - requestPlanChanges: (note?: string) => Promise; + }, dedupeKey: string) => Promise; + approvePlan: (dedupeKey: string) => Promise; + requestPlanChanges: ( + note: string | undefined, + dedupeKey: string, + ) => Promise; } 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 ', ); @@ -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 }; diff --git a/src/types.ts b/src/types.ts index e0276bf..bb69f0b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -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';