242 lines
9.7 KiB
TypeScript
242 lines
9.7 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest';
|
||
import { mkdirSync, mkdtempSync, writeFileSync } from 'fs';
|
||
import { join } from 'path';
|
||
import { tmpdir } from 'os';
|
||
import { Worker } from './worker.js';
|
||
import { userPiecesDir } from './user-folder/paths.js';
|
||
import type { AppConfig } from './config.js';
|
||
import type { Job } from './db/repository.js';
|
||
|
||
/**
|
||
* executeJob の pre-flight ゲート（切り出しリファクタ前のピン留めテスト）。
|
||
*
|
||
* 対象:
|
||
* - MCP 認証ゲート: piece.required_mcp のトークンが無い場合、
|
||
* status='waiting_human' + wait_reason='mcp_auth_required' で park する。
|
||
* wait_reason の永続化が漏れるとジョブが沈黙して永久停止する既知の
|
||
* 失敗モードがあるため、パッチ内容を厳密に固定する。
|
||
* - Model-mismatch requeue ゲート: piece が pin したモデルをこの worker が
|
||
* サーブできない場合、status='queued' + workerId=null で再キューする。
|
||
*
|
||
* どちらも executeJob を実際に走らせ、リポジトリモックへの書き込みを検証する
|
||
* （Given-When-Then）。
|
||
*/
|
||
|
||
function makeConfig(overrides: Partial<AppConfig>): AppConfig {
|
||
return {
|
||
provider: {
|
||
model: 'test-model',
|
||
workers: [{ id: 'worker-1', endpoint: 'http://localhost:11434/v1' }],
|
||
},
|
||
worktreeDir: '/tmp/worker-gate-test',
|
||
concurrency: 1,
|
||
maxMovements: 30,
|
||
retry: { maxAttempts: 3, backoffSeconds: [60, 300, 900] },
|
||
ask: { maxPerJob: 2 },
|
||
subtasks: { maxDepth: 2 },
|
||
tools: {
|
||
searxngUrl: 'http://localhost:8080',
|
||
visionModel: 'vision',
|
||
visionTimeout: 60,
|
||
visionMaxTokens: 1024,
|
||
webfetchTimeout: 30,
|
||
websearchTimeout: 15,
|
||
webfetchAllowedHosts: [],
|
||
},
|
||
...overrides,
|
||
} as AppConfig;
|
||
}
|
||
|
||
function makeLocalTaskJob(overrides: Partial<Job>): Job {
|
||
return {
|
||
id: 'job-gate-1',
|
||
repo: 'local/task-77',
|
||
issueNumber: 77,
|
||
prNumber: null,
|
||
status: 'running',
|
||
pieceName: 'pin-gate-piece',
|
||
requiredRole: 'auto',
|
||
requiredProfile: 'auto',
|
||
currentMovement: null,
|
||
instruction: 'do something',
|
||
branchName: null,
|
||
worktreePath: null,
|
||
attempt: 1,
|
||
maxAttempts: 3,
|
||
nextRetryAt: null,
|
||
errorSummary: null,
|
||
resumeMovement: null,
|
||
askCount: 0,
|
||
workerId: 'worker-1',
|
||
parentJobId: null,
|
||
subtaskDepth: 0,
|
||
createdAt: new Date().toISOString(),
|
||
updatedAt: new Date().toISOString(),
|
||
...overrides,
|
||
} as unknown as Job;
|
||
}
|
||
|
||
function makeRepoMock() {
|
||
const updateJobCalls: Array<[string, Record<string, unknown>]> = [];
|
||
const repo = {
|
||
lockIssue: vi.fn().mockResolvedValue(true),
|
||
unlockIssue: vi.fn().mockResolvedValue(undefined),
|
||
addAuditLog: vi.fn().mockResolvedValue(undefined),
|
||
updateWorkerNodeHealth: vi.fn().mockResolvedValue(undefined),
|
||
updateJob: vi.fn().mockImplementation((id: string, patch: Record<string, unknown>) => {
|
||
updateJobCalls.push([id, patch]);
|
||
return Promise.resolve(undefined);
|
||
}),
|
||
getLocalTask: vi.fn().mockResolvedValue(null),
|
||
updateLocalTask: vi.fn().mockResolvedValue(undefined),
|
||
addLocalTaskComment: vi.fn().mockResolvedValue(undefined),
|
||
};
|
||
return { repo, updateJobCalls };
|
||
}
|
||
|
||
/** Given: userFolderRoot 配下の local オーナー custom piece ディレクトリに piece を配置 */
|
||
function writeCustomPiece(userFolderRoot: string, name: string, extraTopLevel: string): void {
|
||
const dir = userPiecesDir(userFolderRoot, 'local');
|
||
mkdirSync(dir, { recursive: true });
|
||
const yaml = [
|
||
`name: ${name}`,
|
||
'description: gate pinning test piece',
|
||
'initial_movement: respond',
|
||
extraTopLevel,
|
||
'movements:',
|
||
' - name: respond',
|
||
' persona: assistant',
|
||
' instruction: answer the user',
|
||
' default_next: COMPLETE',
|
||
' rules: []',
|
||
'',
|
||
].join('\n');
|
||
writeFileSync(join(dir, `${name}.yaml`), yaml, 'utf-8');
|
||
}
|
||
|
||
describe('executeJob pre-flight gates (behaviour pinning)', () => {
|
||
it('parks the job as waiting_human with wait_reason=mcp_auth_required when a required MCP token is missing', async () => {
|
||
// Given: required_mcp を持つ custom piece と、MCP トークンを一切持たない worker
|
||
const root = mkdtempSync(join(tmpdir(), 'worker-gate-mcp-'));
|
||
const userFolderRoot = join(root, 'users');
|
||
writeCustomPiece(userFolderRoot, 'pin-gate-piece', 'required_mcp:\n - github-mcp');
|
||
const { repo, updateJobCalls } = makeRepoMock();
|
||
const worker = new Worker(
|
||
'worker-1',
|
||
'http://localhost:11434/v1',
|
||
'test-model',
|
||
repo as never,
|
||
makeConfig({ worktreeDir: join(root, 'worktree'), userFolderRoot }),
|
||
);
|
||
|
||
// When: executeJob を最後まで走らせる
|
||
await (worker as unknown as { executeJob: (job: Job) => Promise<void> }).executeJob(
|
||
makeLocalTaskJob({}),
|
||
);
|
||
|
||
// Then: wait_reason='mcp_auth_required' を含む park パッチが永続化される
|
||
//（resume_movement は piece の initial_movement）
|
||
const parkCall = updateJobCalls.find(([, patch]) => patch.status === 'waiting_human');
|
||
expect(parkCall).toBeDefined();
|
||
expect(parkCall![1]).toEqual({
|
||
status: 'waiting_human',
|
||
waitReason: 'mcp_auth_required',
|
||
resumeMovement: 'respond',
|
||
});
|
||
// Then: ユーザーに system イベントコメントで連携を促す
|
||
expect(repo.addLocalTaskComment).toHaveBeenCalledWith(
|
||
77,
|
||
'system',
|
||
expect.stringContaining('github-mcp'),
|
||
'event',
|
||
);
|
||
// Then: 失敗扱いにはならず、issue ロックは finally で必ず解放される
|
||
expect(updateJobCalls.some(([, patch]) => patch.status === 'failed')).toBe(false);
|
||
expect(repo.unlockIssue).toHaveBeenCalledWith('local/task-77', 77);
|
||
});
|
||
|
||
it('requeues the job (status=queued, workerId=null) when the piece pins a model this worker cannot serve', async () => {
|
||
// Given: model を pin した custom piece と、そのモデルを持たない direct worker
|
||
const root = mkdtempSync(join(tmpdir(), 'worker-gate-model-'));
|
||
const userFolderRoot = join(root, 'users');
|
||
writeCustomPiece(userFolderRoot, 'pin-gate-piece', 'model: exotic-model');
|
||
const { repo, updateJobCalls } = makeRepoMock();
|
||
const worker = new Worker(
|
||
'worker-1',
|
||
'http://localhost:11434/v1',
|
||
'test-model',
|
||
repo as never,
|
||
makeConfig({ worktreeDir: join(root, 'worktree'), userFolderRoot }),
|
||
);
|
||
// initialize() 相当: probed available models（exotic-model を含まない）
|
||
(worker as unknown as { availableModels: Set<string> }).availableModels = new Set(['test-model']);
|
||
|
||
// When
|
||
await (worker as unknown as { executeJob: (job: Job) => Promise<void> }).executeJob(
|
||
makeLocalTaskJob({}),
|
||
);
|
||
|
||
// Then: queued に戻し worker を外す（他 worker が拾えるように）
|
||
const requeueCall = updateJobCalls.find(([, patch]) => patch.status === 'queued');
|
||
expect(requeueCall).toBeDefined();
|
||
expect(requeueCall![1]).toEqual({
|
||
status: 'queued',
|
||
workerId: null,
|
||
errorSummary: 'Required model exotic-model is not available on worker-1',
|
||
});
|
||
expect(repo.addAuditLog).toHaveBeenCalledWith('job-gate-1', 'job_requeued_model_mismatch', 'worker', {
|
||
workerId: 'worker-1',
|
||
requiredModel: 'exotic-model',
|
||
availableModels: ['test-model'],
|
||
});
|
||
// Then: park はしない
|
||
expect(updateJobCalls.some(([, patch]) => patch.status === 'waiting_human')).toBe(false);
|
||
expect(repo.unlockIssue).toHaveBeenCalledWith('local/task-77', 77);
|
||
});
|
||
|
||
it('does NOT requeue a model-mismatched job when requiredWorkerId pins it to this worker (pin wins over piece.model)', async () => {
|
||
// Given: 同じ model-mismatch 条件（worker が exotic-model をサーブできない）だが、
|
||
// job が worker-1 に requiredWorkerId でピン留めされている。ピン留めされたジョブは
|
||
// worker-1 にしか claim されないため（Task 8 claim SQL）、ここで requeue すると
|
||
// claim → mismatch → requeue の無限ループになる。
|
||
//
|
||
// 実行経路は requeueOnModelMismatch の直後に startPieceRun（実 LLM 呼び出し込みの
|
||
// ReAct ループ）へ進んでしまい、そこまでモックするのは過剰なため、
|
||
// executeJob から切り出し済みの private ゲートメソッドを直接呼ぶ（既存の
|
||
// 「requeue する」ケースは executeJob 経由で pin-gate-piece 一式ごと検証済みなので、
|
||
// ここでは pin による早期リターンだけを焦点にする）。
|
||
const { repo, updateJobCalls } = makeRepoMock();
|
||
const worker = new Worker(
|
||
'worker-1',
|
||
'http://localhost:11434/v1',
|
||
'test-model',
|
||
repo as never,
|
||
makeConfig({ worktreeDir: '/tmp/worker-gate-pin-model' }),
|
||
);
|
||
(worker as unknown as { availableModels: Set<string> }).availableModels = new Set(['test-model']);
|
||
|
||
const piece = {
|
||
name: 'pin-gate-piece',
|
||
description: 'gate pinning test piece',
|
||
max_movements: 30,
|
||
initial_movement: 'respond',
|
||
model: 'exotic-model',
|
||
movements: [
|
||
{ name: 'respond', persona: 'assistant', instruction: 'answer', rules: [], default_next: 'COMPLETE' },
|
||
],
|
||
};
|
||
|
||
// When: requeueOnModelMismatch を requiredWorkerId='worker-1' の job で直接呼ぶ
|
||
const requeued = await (
|
||
worker as unknown as {
|
||
requeueOnModelMismatch: (job: Job, piece: unknown) => Promise<boolean>;
|
||
}
|
||
).requeueOnModelMismatch(makeLocalTaskJob({ requiredWorkerId: 'worker-1' }), piece);
|
||
|
||
// Then: 「requeue しない」(false) が返り、queued への書き戻し・監査ログのどちらも発生しない
|
||
expect(requeued).toBe(false);
|
||
expect(updateJobCalls.some(([, patch]) => patch.status === 'queued')).toBe(false);
|
||
expect(repo.addAuditLog).not.toHaveBeenCalled();
|
||
});
|
||
});
|