feat: support MCP elicitation and rebuild release

This commit is contained in:
shiyue
2026-08-24 17:39:41 +08:00
parent dd233a40e8
commit bd20a79d4b
69 changed files with 8978 additions and 13 deletions

174
lib/gitea-mcp-probe.js Normal file
View File

@@ -0,0 +1,174 @@
'use strict';
/**
* 对线程级 gitea-mcp stdio 配置做最小 MCP initialize 预检。
*
* 预检只负责判断“命令能否启动并完成协议握手”,不调用 Gitea 工具,也不把
* stderr、Token 或响应正文写入日志。真正的 Agent 工作仍由 Codex App 按线程
* 配置启动另一份 gitea-mcp 进程完成。
*/
const { spawn: defaultSpawn } = require('child_process');
const readline = require('readline');
const DEFAULT_TIMEOUT_MS = 8000;
const MAX_TIMEOUT_MS = 30000;
const MCP_PROTOCOL_VERSION = '2024-11-05';
function safeText(value, max = 512) {
return String(value ?? '').trim().slice(0, max);
}
function probeError(code, message, details = {}) {
return Object.assign(new Error(message), { code, ...details });
}
function normalizeTimeout(value) {
const numeric = Number(value);
if (!Number.isFinite(numeric)) return DEFAULT_TIMEOUT_MS;
return Math.min(MAX_TIMEOUT_MS, Math.max(250, Math.floor(numeric)));
}
function initializeRequest() {
return {
jsonrpc: '2.0',
id: 1,
method: 'initialize',
params: {
protocolVersion: MCP_PROTOCOL_VERSION,
capabilities: {},
clientInfo: { name: 'cc-web-gitea-preflight', version: '1.0.0' },
},
};
}
function isInitializeResponse(message) {
return message
&& message.jsonrpc === '2.0'
&& (message.id === 1 || message.id === '1')
&& (Object.prototype.hasOwnProperty.call(message, 'result')
|| Object.prototype.hasOwnProperty.call(message, 'error'));
}
function sendJsonLine(stdin, value) {
stdin.write(`${JSON.stringify(value)}\n`);
}
/**
* @param {object} options
* @param {string} options.command MCP 命令或绝对路径
* @param {string[]} [options.args] 命令参数
* @param {object} [options.env] 线程级环境变量
* @param {string} [options.cwd] 子进程工作目录
* @param {number} [options.timeoutMs] 握手超时,默认 8 秒
* @param {Function} [options.spawnImpl] 测试替身,签名同 child_process.spawn
* @returns {Promise<{ok:true,protocolVersion:string}>}
*/
function probeGiteaMcp(options = {}) {
const command = safeText(options.command, 2048);
if (!command) return Promise.reject(probeError('gitea_mcp_command_empty', 'Gitea MCP 命令为空。'));
const args = Array.isArray(options.args) ? options.args.map((item) => String(item)) : [];
const timeoutMs = normalizeTimeout(options.timeoutMs);
const spawnImpl = typeof options.spawnImpl === 'function' ? options.spawnImpl : defaultSpawn;
const env = { ...process.env, ...(options.env && typeof options.env === 'object' ? options.env : {}) };
const cwd = safeText(options.cwd, 4096) || process.cwd();
return new Promise((resolve, reject) => {
let child;
let settled = false;
let timer = null;
let sawInvalidLine = false;
const finish = (error, result) => {
if (settled) return;
settled = true;
if (timer) clearTimeout(timer);
try {
if (child?.stdin && !child.stdin.destroyed) child.stdin.end();
} catch {}
try {
if (child && !child.killed) child.kill('SIGTERM');
} catch {}
if (error) reject(error);
else resolve(result);
};
try {
child = spawnImpl(command, args, {
cwd,
env,
stdio: ['pipe', 'pipe', 'pipe'],
windowsHide: true,
});
} catch (error) {
finish(probeError('gitea_mcp_spawn_failed', 'Gitea MCP 进程无法启动。', { cause: error }));
return;
}
const handleLine = (line) => {
const text = String(line || '').trim();
if (!text) return;
let message;
try {
message = JSON.parse(text);
} catch {
// stdio 协议要求 JSONL兼容少数实现的启动提示最终仍以超时/退出分类。
sawInvalidLine = true;
return;
}
if (!isInitializeResponse(message)) return;
if (message.error) {
finish(probeError('gitea_mcp_handshake_failed', 'Gitea MCP initialize 握手失败。', {
rpcCode: message.error.code ?? null,
}));
return;
}
const result = message.result && typeof message.result === 'object' ? message.result : {};
try {
// MCP 客户端在收到 initialize 响应后必须发送 initialized 通知。
sendJsonLine(child.stdin, { jsonrpc: '2.0', method: 'notifications/initialized', params: {} });
} catch (error) {
finish(probeError('gitea_mcp_handshake_failed', 'Gitea MCP initialized 通知发送失败。', { cause: error }));
return;
}
finish(null, {
ok: true,
protocolVersion: safeText(result.protocolVersion, 64) || MCP_PROTOCOL_VERSION,
});
};
if (child?.stdout) {
const lineReader = readline.createInterface({ input: child.stdout });
lineReader.on('line', handleLine);
child.once?.('close', () => lineReader.close());
}
child?.once?.('error', (error) => {
finish(probeError('gitea_mcp_spawn_failed', 'Gitea MCP 进程启动失败。', { cause: error }));
});
child?.once?.('exit', (code, signal) => {
if (settled) return;
finish(probeError('gitea_mcp_start_failed', 'Gitea MCP 进程在握手前退出。', {
exitCode: code ?? null,
signal: signal || null,
invalidOutput: sawInvalidLine,
}));
});
timer = setTimeout(() => {
finish(probeError('gitea_mcp_handshake_timeout', `Gitea MCP initialize 握手超时(${timeoutMs}ms`));
}, timeoutMs);
try {
sendJsonLine(child.stdin, initializeRequest());
} catch (error) {
finish(probeError('gitea_mcp_handshake_failed', 'Gitea MCP initialize 请求发送失败。', { cause: error }));
}
});
}
module.exports = {
DEFAULT_TIMEOUT_MS,
MAX_TIMEOUT_MS,
MCP_PROTOCOL_VERSION,
probeGiteaMcp,
};

View File

@@ -0,0 +1,7 @@
'use strict';
// 兼容聚合入口Webhook 与工作区各自保持独立实现,集成方可按需引用一个模块。
module.exports = {
...require('./gitea-webhook'),
...require('./gitea-workspace'),
};

493
lib/gitea-webhook.js Normal file
View File

@@ -0,0 +1,493 @@
'use strict';
const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const { parseWorkflowMarker } = require('./gitea-workflow-codex');
const DEFAULT_MAX_BODY_BYTES = 2 * 1024 * 1024;
const DEFAULT_BOT_LOGIN = 'ccweb-bot';
function cloneJson(value) {
if (value === undefined) return undefined;
return JSON.parse(JSON.stringify(value));
}
function safeString(value) {
return typeof value === 'string' ? value.trim() : '';
}
function escapeRegExp(value) {
return String(value).replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
}
function timingSafeHexEqual(expectedHex, receivedHex) {
const expected = Buffer.from(String(expectedHex || '').toLowerCase(), 'hex');
const received = Buffer.from(String(receivedHex || '').toLowerCase(), 'hex');
return expected.length > 0
&& expected.length === received.length
&& crypto.timingSafeEqual(expected, received);
}
function normalizeSignature(value) {
const raw = safeString(value).replace(/^sha256[=:]/i, '').trim();
return /^[0-9a-f]{64}$/i.test(raw) ? raw.toLowerCase() : '';
}
function verifyWebhookSignature(rawBody, signature, secret) {
const configuredSecret = String(secret || '');
if (!configuredSecret) return false;
const expected = crypto.createHmac('sha256', configuredSecret)
.update(Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(String(rawBody || '')))
.digest('hex');
return timingSafeHexEqual(expected, normalizeSignature(signature));
}
function getHeader(headers, names) {
const source = headers || {};
for (const name of names) {
const value = source[name] ?? source[name.toLowerCase()];
if (Array.isArray(value)) return safeString(value[0]);
if (value !== undefined) return safeString(value);
}
const lowerNames = new Set(names.map((name) => name.toLowerCase()));
for (const [name, value] of Object.entries(source)) {
if (!lowerNames.has(name.toLowerCase())) continue;
return Array.isArray(value) ? safeString(value[0]) : safeString(value);
}
return '';
}
function atomicWriteJson(filePath, value) {
const target = path.resolve(filePath);
fs.mkdirSync(path.dirname(target), { recursive: true });
const tempPath = `${target}.${process.pid}.${crypto.randomBytes(5).toString('hex')}.tmp`;
fs.writeFileSync(tempPath, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 });
try {
fs.renameSync(tempPath, target);
} finally {
try { if (fs.existsSync(tempPath)) fs.unlinkSync(tempPath); } catch {}
}
}
class DeliveryStore {
constructor(options = {}) {
this.filePath = options.filePath ? path.resolve(options.filePath) : '';
this.maxEntries = Number.isSafeInteger(options.maxEntries) && options.maxEntries > 0
? options.maxEntries : 10_000;
this.entries = new Map();
this.load();
}
load() {
if (!this.filePath || !fs.existsSync(this.filePath)) return;
try {
const parsed = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
const source = parsed && typeof parsed === 'object' && !Array.isArray(parsed)
? parsed.entries || parsed : {};
for (const [key, value] of Object.entries(source)) {
if (key && value && typeof value === 'object') this.entries.set(key, value);
}
} catch {
// 损坏的去重文件不应阻止服务启动;新请求会覆盖为可恢复的合法文件。
this.entries.clear();
}
}
get(key) {
const value = this.entries.get(String(key));
return value ? cloneJson(value) : null;
}
has(key) {
return this.entries.has(String(key));
}
put(key, value) {
const normalizedKey = String(key);
this.entries.set(normalizedKey, cloneJson(value));
while (this.entries.size > this.maxEntries) {
const oldest = this.entries.keys().next().value;
this.entries.delete(oldest);
}
if (this.filePath) {
atomicWriteJson(this.filePath, {
schemaVersion: 1,
entries: Object.fromEntries(this.entries),
});
}
}
claimDelivery(key, value = {}) {
const existing = this.get(key);
if (existing) return { duplicate: true, record: existing };
this.put(key, value);
return { duplicate: false, record: this.get(key) };
}
getDelivery(key) { return this.get(key); }
}
function repositoryFromPayload(payload) {
const repository = payload?.repository || payload?.repo || {};
const owner = repository.owner || repository.organization || {};
const ownerName = typeof owner === 'string'
? safeString(owner)
: safeString(owner.login || owner.username || owner.name || repository.owner_name);
const repoName = safeString(repository.name || repository.repo_name);
const fullName = safeString(repository.full_name || (ownerName && repoName ? `${ownerName}/${repoName}` : ''));
const parts = fullName.split('/').filter(Boolean);
const normalizedOwner = ownerName || parts.slice(0, -1).join('/');
const normalizedRepo = repoName || parts.at(-1) || '';
if (!normalizedOwner || !normalizedRepo || normalizedOwner.includes('/') || normalizedRepo.includes('/')) {
return null;
}
return {
owner: normalizedOwner,
name: normalizedRepo,
fullName: `${normalizedOwner}/${normalizedRepo}`,
cloneUrl: safeString(repository.clone_url || repository.cloneUrl || repository.html_url || ''),
defaultBranch: safeString(repository.default_branch || repository.defaultBranch || ''),
private: repository.private === true,
};
}
function resourceFromPayload(payload, eventName) {
const issue = payload?.issue || payload?.pull_request || payload?.pullRequest || {};
const pullRequest = payload?.pull_request || payload?.pullRequest || issue?.pull_request;
const number = Number(issue.number || payload?.number || pullRequest?.number);
if (!Number.isSafeInteger(number) || number <= 0) return null;
const isPullRequest = eventName.includes('pull_request')
|| !!issue.pull_request
|| !!payload?.pull_request
|| !!payload?.pullRequest;
const kind = isPullRequest ? 'pull_request' : 'issue';
return {
kind,
number,
key: `${kind}:${number}`,
title: safeString(issue.title || pullRequest?.title),
baseRef: safeString(pullRequest?.base?.ref || pullRequest?.base_ref),
headRef: safeString(pullRequest?.head?.ref || pullRequest?.head_ref),
headRepo: safeString(pullRequest?.head?.repo?.full_name || pullRequest?.head_repo),
};
}
function commentFromPayload(payload) {
const comment = payload?.comment || payload?.review?.comment || {};
const body = typeof comment.body === 'string'
? comment.body
: (typeof payload?.comment_body === 'string' ? payload.comment_body : '');
if (!body) return null;
const author = comment.user || comment.author || payload?.sender || payload?.user || {};
return {
id: String(comment.id || comment.number || payload?.comment_id || '').trim(),
body,
author: {
id: author.id ?? null,
login: safeString(author.login || author.username || author.name),
type: safeString(author.type),
},
createdAt: safeString(comment.created_at || comment.createdAt || payload?.created_at),
updatedAt: safeString(comment.updated_at || comment.updatedAt || payload?.updated_at),
};
}
function normalizeEventName(headerValue, payload) {
const header = safeString(headerValue).toLowerCase().replace(/[-\s]/g, '_');
const action = safeString(payload?.action).toLowerCase();
if (['issue_comment', 'issues_comment', 'pull_request_comment', 'pullrequest_comment'].includes(header)) {
return header.includes('pull') ? 'pull_request_comment' : 'issue_comment';
}
// 某些 Gitea 版本把普通 Issue 评论复用为 issues + comment 字段;
// 只有存在 comment 时才归一化,避免新建 Issue 被误触发。
if (header === 'issues' && (action === 'comment' || payload?.comment)) return 'issue_comment';
if (header === 'pull_request' && (['comment', 'review_comment'].includes(action) || payload?.comment)) return 'pull_request_comment';
return header;
}
function parseBotMention(body, botLogin = DEFAULT_BOT_LOGIN) {
const login = safeString(botLogin) || DEFAULT_BOT_LOGIN;
const mention = new RegExp(`(^|[^\\w@])@${escapeRegExp(login)}(?=$|[^\\w-])`, 'i');
const match = mention.exec(String(body || ''));
if (!match) return null;
const before = String(body).slice(0, match.index);
const after = String(body).slice(match.index + match[0].length);
const instruction = `${before}${after}`
.replace(/^[\s:,,、-]+/, '')
.replace(/[\s]+$/, '')
.trim();
return {
botLogin: login,
mention: `@${login}`,
instruction,
index: match.index + match[1].length,
};
}
function isBotAuthor(author, botIdentity = {}) {
const login = safeString(author?.login).toLowerCase();
const configuredLogin = safeString(botIdentity.login || botIdentity.username || DEFAULT_BOT_LOGIN).toLowerCase();
if (login && configuredLogin && login === configuredLogin) return true;
if (botIdentity.id !== undefined && botIdentity.id !== null && author?.id !== null && author?.id !== undefined) {
return String(botIdentity.id) === String(author.id);
}
return false;
}
function normalizeWebhookEvent({ payload, headers = {}, instanceId = 'default', botIdentity }) {
if (!payload || typeof payload !== 'object' || Array.isArray(payload)) {
return { ignored: true, reason: 'invalid_payload' };
}
const eventName = normalizeEventName(getHeader(headers, ['x-gitea-event', 'x-event-key', 'x-github-event']), payload);
if (!['issue_comment', 'pull_request_comment'].includes(eventName)) {
return { ignored: true, reason: 'unsupported_event', eventName };
}
const comment = commentFromPayload(payload);
if (!comment) return { ignored: true, reason: 'not_comment', eventName };
// 带完整关联标识的 cc-web 回帖即使 Gitea 版本未携带作者字段,也不得再次触发。
if (parseWorkflowMarker(comment.body)) {
return { ignored: true, reason: 'workflow_marker_comment', eventName, commentId: comment.id || null };
}
if (isBotAuthor(comment.author, botIdentity)) {
return { ignored: true, reason: 'bot_self_comment', eventName, commentId: comment.id || null };
}
const mention = parseBotMention(comment.body, botIdentity?.login || DEFAULT_BOT_LOGIN);
if (!mention) return { ignored: true, reason: 'missing_mention', eventName, commentId: comment.id || null };
const repository = repositoryFromPayload(payload);
if (!repository) return { ignored: true, reason: 'repository_unresolved', eventName, commentId: comment.id || null };
const resource = resourceFromPayload(payload, eventName);
if (!resource) return { ignored: true, reason: 'resource_unresolved', eventName, commentId: comment.id || null };
const sender = payload.sender || payload.user || null;
return {
ignored: false,
eventName,
eventType: eventName,
instanceId: safeString(instanceId) || 'default',
deliveryId: getHeader(headers, ['x-gitea-delivery', 'x-delivery-id', 'x-github-delivery']) || safeString(payload.delivery_id),
repository,
repo: repository,
resource,
refs: { base: resource.baseRef || null, head: resource.headRef || null },
comment,
mention,
sender,
actor: {
login: safeString(sender?.login || sender?.username || sender?.name),
id: sender?.id ?? null,
isBot: sender?.is_bot === true || String(sender?.type || '').toLowerCase() === 'bot',
},
receivedAt: new Date().toISOString(),
};
}
function createTaskId(deliveryKey) {
return `gitea-task-${crypto.createHash('sha256').update(String(deliveryKey)).digest('hex').slice(0, 24)}`;
}
function readRawBody(req, maxBytes = DEFAULT_MAX_BODY_BYTES) {
return new Promise((resolve, reject) => {
const chunks = [];
let total = 0;
let settled = false;
const fail = (error) => {
if (settled) return;
settled = true;
reject(error);
};
req.on('data', (chunk) => {
if (settled) return;
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
total += buffer.length;
if (total > maxBytes) {
fail(Object.assign(new Error('Webhook 请求体过大。'), { code: 'body_too_large' }));
try { req.destroy(); } catch {}
return;
}
chunks.push(buffer);
});
req.on('end', () => {
if (settled) return;
settled = true;
resolve(Buffer.concat(chunks));
});
req.on('error', fail);
});
}
function readDeliveryRecord(store, key) {
if (typeof store?.get === 'function') return store.get(key);
if (typeof store?.getDelivery === 'function') return store.getDelivery(key);
return null;
}
function claimDeliveryRecord(store, key, record) {
if (typeof store?.claimDelivery === 'function') {
return store.claimDelivery(key, record);
}
if (typeof store?.put === 'function') {
store.put(key, record);
return { duplicate: false, record };
}
throw new TypeError('deliveryStore 必须提供 get/put 或 getDelivery/claimDelivery。');
}
function createGiteaWebhookReceiver(options = {}) {
let secret = options.secret ?? process.env.CC_WEB_GITEA_WEBHOOK_SECRET ?? '';
let botToken = options.botToken ?? '';
const instanceId = safeString(options.instanceId || process.env.CC_WEB_GITEA_INSTANCE_ID || process.env.GITEA_INSTANCE_ID) || 'default';
const botIdentity = {
login: safeString(options.botLogin || process.env.CC_WEB_GITEA_BOT_LOGIN || process.env.GITEA_BOT_LOGIN) || DEFAULT_BOT_LOGIN,
id: options.botId ?? process.env.CC_WEB_GITEA_BOT_ID ?? process.env.GITEA_BOT_ID ?? null,
};
const maxBodyBytes = Number.isSafeInteger(options.maxBodyBytes) && options.maxBodyBytes > 0
? options.maxBodyBytes : DEFAULT_MAX_BODY_BYTES;
const deliveryStore = options.deliveryStore || new DeliveryStore({
filePath: options.deliveryFile || process.env.CC_WEB_GITEA_DELIVERY_FILE,
});
const onTask = typeof options.onTask === 'function' ? options.onTask : async () => {};
const inflight = new Set();
async function processWebhook({ headers = {}, rawBody }) {
const body = Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(String(rawBody || ''));
if (options.requireBotToken && !String(botToken || '').trim()) {
return { statusCode: 503, payload: { ok: false, error: 'gitea_workflow_not_configured' } };
}
const signature = getHeader(headers, ['x-gitea-signature', 'x-hub-signature-256', 'x-signature']);
if (secret && !verifyWebhookSignature(body, signature, secret)) {
return { statusCode: 401, payload: { ok: false, error: 'invalid_signature' } };
}
let payload;
try {
payload = JSON.parse(body.toString('utf8'));
} catch {
return { statusCode: 400, payload: { ok: false, error: 'invalid_json' } };
}
const deliveryId = getHeader(headers, ['x-gitea-delivery', 'x-delivery-id', 'x-github-delivery'])
|| safeString(payload.delivery_id);
if (!deliveryId) return { statusCode: 400, payload: { ok: false, error: 'missing_delivery_id' } };
const deliveryKey = `${instanceId}:${deliveryId}`;
const existing = readDeliveryRecord(deliveryStore, deliveryKey);
if (existing) {
return {
statusCode: 200,
payload: { ok: true, duplicate: true, status: 'duplicate', deliveryId, taskId: existing.taskId || null, reason: existing.reason || null },
record: existing,
};
}
const event = normalizeWebhookEvent({ payload, headers, instanceId, botIdentity });
event.deliveryId = deliveryId;
const record = {
deliveryKey,
deliveryId,
instanceId,
taskId: event.ignored ? null : createTaskId(deliveryKey),
status: event.ignored ? 'ignored' : 'accepted',
reason: event.reason || null,
eventName: event.eventName || null,
receivedAt: event.receivedAt || new Date().toISOString(),
};
const claimed = claimDeliveryRecord(deliveryStore, deliveryKey, record);
if (claimed?.duplicate) {
const duplicateRecord = claimed.record || readDeliveryRecord(deliveryStore, deliveryKey) || record;
return {
statusCode: 200,
payload: { ok: true, duplicate: true, status: 'duplicate', deliveryId, taskId: duplicateRecord.taskId || null },
record: duplicateRecord,
};
}
if (event.ignored) {
return { statusCode: 200, payload: { ok: true, ignored: true, status: 'ignored', reason: event.reason }, record, event };
}
const task = {
taskId: record.taskId,
deliveryKey,
deliveryId,
instanceId,
event,
receivedAt: record.receivedAt,
};
// Webhook 先快速确认,工作区 clone/fetch 等副作用由上层异步处理。
const taskPromise = Promise.resolve().then(() => onTask(cloneJson(task))).catch((error) => {
if (typeof options.onTaskError === 'function') {
try { options.onTaskError(error, task); } catch {}
}
}).finally(() => inflight.delete(taskPromise));
inflight.add(taskPromise);
return {
statusCode: 202,
payload: { ok: true, accepted: true, state: 'queued', status: 'accepted', deliveryId, taskId: record.taskId },
record,
event,
task,
};
}
async function handle(req, res) {
let result;
try {
const rawBody = await readRawBody(req, maxBodyBytes);
result = await processWebhook({ headers: req.headers || {}, rawBody });
} catch (error) {
const statusCode = error?.code === 'body_too_large' ? 413 : 400;
result = { statusCode, payload: { ok: false, error: error?.code || 'invalid_request' } };
}
if (!res.headersSent) {
res.writeHead(result.statusCode, {
'Content-Type': 'application/json; charset=utf-8',
'Cache-Control': 'no-cache',
});
res.end(JSON.stringify(result.payload));
}
return result;
}
async function drain() {
await Promise.all(Array.from(inflight));
}
function updateConfig(next = {}) {
if (Object.prototype.hasOwnProperty.call(next, 'secret')) secret = String(next.secret || '');
if (Object.prototype.hasOwnProperty.call(next, 'botToken')) botToken = String(next.botToken || '');
if (Object.prototype.hasOwnProperty.call(next, 'botLogin')) {
botIdentity.login = safeString(next.botLogin) || DEFAULT_BOT_LOGIN;
}
if (Object.prototype.hasOwnProperty.call(next, 'botId')) botIdentity.id = next.botId ?? null;
return {
webhookSecretConfigured: Boolean(secret),
botTokenConfigured: Boolean(botToken),
botLogin: botIdentity.login,
botId: botIdentity.id,
};
}
return { handle, processWebhook, drain, updateConfig, deliveryStore, instanceId, botIdentity };
}
function createGiteaWebhookRoute(receiver, options = {}) {
if (!receiver || typeof receiver.handle !== 'function') {
throw new TypeError('receiver.handle 必须是函数。');
}
const endpoint = String(options.path || '/api/gitea/webhook');
return function handleGiteaWebhookRoute(req, res, urlLike) {
let pathname = typeof urlLike === 'string' ? urlLike : urlLike?.pathname;
if (!pathname) {
try { pathname = new URL(req.url || '/', 'http://localhost').pathname; } catch { pathname = ''; }
}
if (req.method !== 'POST' || pathname !== endpoint) return false;
receiver.handle(req, res);
return true;
};
}
module.exports = {
DEFAULT_MAX_BODY_BYTES,
DeliveryStore,
createDeliveryStore: (options) => new DeliveryStore(options),
createGiteaWebhookRoute,
createGiteaWebhookReceiver,
createTaskId,
normalizeWebhookEvent,
parseBotMention,
verifyWebhookSignature,
};

638
lib/gitea-workflow-codex.js Normal file
View File

@@ -0,0 +1,638 @@
'use strict';
/**
* Gitea Webhook 工作流与 Codex App 之间的薄适配层。
*
* 该模块不负责队列或业务存储,只维护一次 task/turn 的协议状态,方便
* server.js、持久化层和 Gitea REST/MCP 客户端分别注入实现。所有外部副作用
* 都通过 callbacks 注入,因而可以用协议 mock 做确定性回归。
*/
const DEFAULT_GITEA_MCP_COMMAND = 'gitea-mcp';
const WORKFLOW_MARKER_NAME = 'ccweb-gitea';
const WORKFLOW_MARKER_VERSION = '1';
const DEFAULT_REPLY_RETRY_LIMIT = 1;
const WORKFLOW_STATES = Object.freeze({
QUEUED: 'queued',
RUNNING: 'running',
WAITING_USER: 'waiting_user',
VERIFYING_REPLY: 'verifying_reply',
RETRYING_REPLY: 'retrying_reply',
SUCCEEDED: 'succeeded',
SUCCEEDED_WITH_REST_FALLBACK: 'succeeded_with_rest_fallback',
FAILED: 'failed',
FAILED_REPLY: 'failed_reply',
});
const TURN_STATES = Object.freeze({
STARTED: 'started',
COMPLETED: 'completed',
WAITING_USER: 'waiting_user',
FAILED: 'failed',
INTERRUPTED: 'interrupted',
});
function cleanText(value, max = 512) {
const text = String(value ?? '').trim();
return text.length > max ? text.slice(0, max) : text;
}
function requiredText(value, field, max = 512) {
const text = cleanText(value, max);
if (!text) throw new TypeError(`${field} 不能为空。`);
return text;
}
function clone(value) {
if (value === undefined) return undefined;
return JSON.parse(JSON.stringify(value));
}
function isoNow(now) {
const value = typeof now === 'function' ? now() : now;
const date = value instanceof Date ? value : new Date(value || Date.now());
return Number.isNaN(date.getTime()) ? new Date().toISOString() : date.toISOString();
}
function normalizeHost(value) {
const host = requiredText(value, 'Gitea host', 2048).replace(/\/+$/, '');
let parsed;
try {
parsed = new URL(host);
} catch {
throw new TypeError('Gitea host 必须是绝对 URL。');
}
if (!['http:', 'https:'].includes(parsed.protocol) || parsed.username || parsed.password) {
throw new TypeError('Gitea host 只允许 http/https URL不能携带凭据。');
}
return parsed.toString().replace(/\/+$/, '');
}
function buildGiteaMcpServerConfig(options = {}) {
const host = normalizeHost(options.host || options.giteaHost);
const accessToken = requiredText(options.accessToken || options.token, 'Gitea access token', 4096);
const command = cleanText(options.command || DEFAULT_GITEA_MCP_COMMAND, 512);
if (!command) throw new TypeError('gitea-mcp command 不能为空。');
const extraArgs = Array.isArray(options.extraArgs)
? options.extraArgs.map((item) => String(item))
: [];
const config = {
type: 'stdio',
command,
args: ['-t', 'stdio', '-H', host, ...extraArgs],
// host 同时放在标准环境变量中,便于不同版本的官方 gitea-mcp 兼容;
// token 只存在于本线程 config不写入 app-server 进程环境。
env: {
GITEA_HOST: host,
GITEA_ACCESS_TOKEN: accessToken,
},
};
if (Number.isFinite(options.startupTimeoutSec)) {
config.startup_timeout_sec = Math.max(1, Number(options.startupTimeoutSec));
}
if (Number.isFinite(options.toolTimeoutSec)) {
config.tool_timeout_sec = Math.max(1, Number(options.toolTimeoutSec));
}
return config;
}
// 兼容工作流核心服务使用的旧命名;新代码优先使用
// buildGiteaMcpServerConfig/buildGiteaThreadConfig。
function buildGiteaMcpConfig(config = {}, options = {}) {
const source = config.gitea && typeof config.gitea === 'object' ? config.gitea : config;
return buildGiteaMcpServerConfig({
host: options.host || source.host,
accessToken: options.token || options.accessToken || source.token || source.accessToken,
command: options.command || source.mcpCommand || DEFAULT_GITEA_MCP_COMMAND,
extraArgs: options.args && Array.isArray(options.args)
? options.args.filter((item, index) => !(index === 0 && item === '-t') && !(index === 1 && item === 'stdio'))
: [],
});
}
function buildGiteaThreadConfig(options = {}) {
const cwd = requiredText(options.cwd || options.workspacePath, '工作区 cwd', 4096);
const gitea = buildGiteaMcpServerConfig(options);
return {
cwd,
config: {
'mcp_servers.gitea': gitea,
},
// 便于调用方在需要时直接合并到 thread/start 参数server.js 使用 config。
mcpServer: gitea,
};
}
function mergeGiteaThreadConfig(baseConfig = {}, options = {}) {
const thread = buildGiteaThreadConfig(options);
return {
...baseConfig,
...thread.config,
};
}
function extractTurnId(value = {}) {
if (typeof value === 'string' || typeof value === 'number') return cleanText(value, 256) || null;
if (!value || typeof value !== 'object') return null;
const params = value.params && typeof value.params === 'object' ? value.params : value;
const candidates = [
params.turnId,
params.turn_id,
params.turn?.id,
params.item?.turnId,
params.item?.turn_id,
params.item?.turn?.id,
params.event?.turnId,
];
for (const candidate of candidates) {
const turnId = cleanText(candidate, 256);
if (turnId) return turnId;
}
return null;
}
function extractThreadId(value = {}) {
if (typeof value === 'string' || typeof value === 'number') return cleanText(value, 256) || null;
if (!value || typeof value !== 'object') return null;
const params = value.params && typeof value.params === 'object' ? value.params : value;
const candidates = [
params.threadId,
params.thread_id,
params.thread?.id,
params.item?.threadId,
params.item?.thread_id,
];
for (const candidate of candidates) {
const threadId = cleanText(candidate, 256);
if (threadId) return threadId;
}
return null;
}
function escapeMarkerValue(value) {
return String(value ?? '')
.replaceAll('&', '&amp;')
.replaceAll('"', '&quot;')
.replaceAll('<', '&lt;')
.replaceAll('>', '&gt;');
}
function unescapeMarkerValue(value) {
return String(value ?? '')
.replaceAll('&quot;', '"')
.replaceAll('&gt;', '>')
.replaceAll('&lt;', '<')
.replaceAll('&amp;', '&');
}
function buildWorkflowMarker(input = {}) {
const taskId = requiredText(input.taskId, 'taskId', 256);
const turnId = requiredText(input.turnId, 'turnId', 256);
const resourceKey = requiredText(input.resourceKey, 'resourceKey', 1024);
const kind = cleanText(input.kind || 'final', 64) || 'final';
return `<!-- ${WORKFLOW_MARKER_NAME} v="${WORKFLOW_MARKER_VERSION}" taskId="${escapeMarkerValue(taskId)}" turnId="${escapeMarkerValue(turnId)}" resourceKey="${escapeMarkerValue(resourceKey)}" kind="${escapeMarkerValue(kind)}" -->`;
}
function parseWorkflowMarker(body) {
const text = String(body || '');
const markerMatch = text.match(new RegExp(`<!--\\s*${WORKFLOW_MARKER_NAME}\\b([\\s\\S]*?)-->`, 'i'));
if (!markerMatch) return null;
const attrs = {};
const attrPattern = /([a-z][\w-]*)\s*=\s*"([^"]*)"/gi;
let match;
while ((match = attrPattern.exec(markerMatch[1]))) attrs[match[1]] = unescapeMarkerValue(match[2]);
if (attrs.v !== WORKFLOW_MARKER_VERSION) return null;
if (!attrs.taskId || !attrs.turnId || !attrs.resourceKey) return null;
return {
version: attrs.v,
taskId: attrs.taskId,
turnId: attrs.turnId,
resourceKey: attrs.resourceKey,
kind: attrs.kind || 'final',
};
}
function botCandidate(comment = {}) {
return comment.user || comment.author || comment.sender || comment.actor || comment.owner || {};
}
function normalizeBotIdentity(identity = {}) {
if (typeof identity === 'string') return { login: cleanText(identity, 256).toLowerCase() };
return {
id: identity.id === undefined || identity.id === null ? null : String(identity.id),
login: cleanText(identity.login || identity.username || identity.name, 256).toLowerCase() || null,
};
}
function isBotComment(comment = {}, identity = {}) {
const candidate = botCandidate(comment);
const bot = normalizeBotIdentity(identity);
if (candidate.isBot === true || candidate.is_bot === true || String(candidate.type || '').toLowerCase() === 'bot') return true;
if (bot.id && candidate.id !== undefined && candidate.id !== null && String(candidate.id) === bot.id) return true;
const login = cleanText(candidate.login || candidate.username || candidate.name, 256).toLowerCase();
return !!(bot.login && login && login === bot.login);
}
function normalizeComment(comment = {}) {
const candidate = botCandidate(comment);
const resource = comment.resource || comment.issue || comment.pull_request || {};
return {
id: comment.id ?? comment.commentId ?? null,
body: String(comment.body || comment.content || ''),
resourceKey: cleanText(comment.resourceKey || resource.resourceKey, 1024) || null,
user: {
id: candidate.id ?? null,
login: cleanText(candidate.login || candidate.username || candidate.name, 256) || null,
isBot: candidate.isBot === true || candidate.is_bot === true || String(candidate.type || '').toLowerCase() === 'bot',
},
raw: comment,
};
}
function matchWorkflowReply(comment, expected = {}, botIdentity = {}) {
const normalized = normalizeComment(comment);
if (!isBotComment(comment, botIdentity)) return false;
const marker = parseWorkflowMarker(normalized.body);
if (!marker) return false;
if (!['final', 'fallback'].includes(String(marker.kind || 'final').toLowerCase())) return false;
const expectedTurnIds = Array.isArray(expected.turnIds)
? expected.turnIds.map((value) => cleanText(value, 256)).filter(Boolean)
: [cleanText(expected.turnId, 256)].filter(Boolean);
return marker.taskId === cleanText(expected.taskId, 256)
&& expectedTurnIds.includes(marker.turnId)
&& marker.resourceKey === cleanText(expected.resourceKey, 1024);
}
function findWorkflowReply(comments, expected = {}, botIdentity = {}) {
if (!Array.isArray(comments)) return null;
return comments.find((comment) => matchWorkflowReply(comment, expected, botIdentity)) || null;
}
function normalizeVerification(result, expected, botIdentity) {
if (Array.isArray(result)) {
const comment = findWorkflowReply(result, expected, botIdentity);
return comment ? { status: 'confirmed', comment } : { status: 'missing', comment: null };
}
if (result && typeof result === 'object') {
if (Array.isArray(result.comments)) return normalizeVerification(result.comments, expected, botIdentity);
if (result.status === 'unknown' || result.unknown === true) return { status: 'unknown', comment: null };
if (result.confirmed === true || result.found === true || result.accepted === true) {
if (result.comment && !matchWorkflowReply(result.comment, expected, botIdentity)) {
return { status: 'missing', comment: null };
}
return { status: 'confirmed', comment: result.comment || null, commentId: result.commentId || result.id || null };
}
if (result.error || result.status === 'error') return { status: 'unknown', error: result.error || result.message || 'reply verification failed' };
}
if (result === true) return { status: 'confirmed', comment: null };
return { status: 'missing', comment: null };
}
function isWaitingUserEvent(notification = {}) {
const method = String(notification.method || notification.type || '').toLowerCase();
const params = notification.params && typeof notification.params === 'object' ? notification.params : notification;
if (method.includes('requestuserinput') || method.includes('waiting_user') || method.includes('waitinguser')) return true;
return String(params.status || params.turn?.status || params.reason || '').toLowerCase() === 'waiting_user';
}
function buildWaitingUserPrompt(comment = {}) {
const body = String(comment.body || comment.content || '').trim();
return body || '用户已继续评论,请读取本条评论并继续处理。';
}
function buildReplyRetryPrompt(task, finalText, marker) {
return [
'[ccweb-gitea hidden reply retry]',
'只补发上一轮最终回执,不要重复修改代码、执行写操作或重新研究。',
`resourceKey=${task.resourceKey}`,
`taskId=${task.taskId}`,
`turnId=${marker.turnId}`,
'上一轮最终文本:',
String(finalText || '').trim(),
marker.text,
].join('\n');
}
function encodeMarker(input = {}) {
return buildWorkflowMarker({
taskId: input.taskId,
turnId: input.turnId,
resourceKey: input.resourceKey,
kind: input.kind || input.state || 'final',
});
}
function buildWorkflowPrompt(task = {}) {
const instruction = String(task.instruction || task.prompt || '').trim();
const resource = String(task.resourceKey || '').trim();
const workspace = String(task.workspacePath || task.cwd || '').trim();
const marker = task.turnId && task.taskId && resource
? encodeMarker({ taskId: task.taskId, turnId: task.turnId, resourceKey: resource, state: 'final' })
: '';
const waitingMarker = task.turnId && task.taskId && resource
? encodeMarker({ taskId: task.taskId, turnId: task.turnId, resourceKey: resource, kind: 'waiting_user' })
: '';
return [
'你正在执行 cc-web Gitea Workflow 任务。',
'请在当前工作区完成用户请求Gitea 是唯一主交互入口。',
resource ? `资源:${resource}` : '',
workspace ? `工作区:${workspace}` : '',
instruction ? `用户指令:\n${instruction}` : '用户未提供额外指令,请先读取资源上下文。',
'完成后必须使用官方 gitea-mcp 在原 Issue/PR 回帖;最终摘要正文末尾保留 final 隐藏标记。',
'如果需要用户补充信息或遇到阻塞,请只在原 Issue/PR 提问并在正文末尾保留 waiting_user 隐藏标记,然后停止,不要继续修改。',
`final 隐藏标记:${marker}`,
`waiting_user 隐藏标记:${waitingMarker}`,
].filter(Boolean).join('\n\n');
}
function classifyReplyComments(comments, expected = {}, botIdentity = {}) {
const verification = normalizeVerification(comments, expected, botIdentity);
return {
status: verification.status,
comment: verification.comment || null,
commentId: verification.commentId || verification.comment?.id || null,
error: verification.error || null,
};
}
function createGiteaWorkflowCodex(options = {}) {
const now = options.now || (() => new Date());
const botIdentity = normalizeBotIdentity(options.botIdentity || options.botLogin || {});
const replyRetryLimit = Number.isFinite(options.replyRetryLimit)
? Math.max(0, Math.floor(options.replyRetryLimit))
: DEFAULT_REPLY_RETRY_LIMIT;
const tasks = new Map();
const sessions = new Map();
function emit(task, event, extra = {}) {
task.updatedAt = isoNow(now);
const snapshot = { ...clone(task), event, ...clone(extra) };
if (typeof options.onStateChange === 'function') options.onStateChange(snapshot);
return snapshot;
}
function createTask(input = {}) {
const taskId = requiredText(input.taskId, 'taskId', 256);
const sessionKey = requiredText(input.sessionKey, 'sessionKey', 1024);
const resourceKey = requiredText(input.resourceKey, 'resourceKey', 1024);
const current = tasks.get(taskId);
if (current) return clone(current);
const task = {
taskId,
sessionKey,
resourceKey,
threadId: cleanText(input.threadId, 256) || null,
parentTaskId: cleanText(input.parentTaskId, 256) || null,
state: input.state || WORKFLOW_STATES.QUEUED,
turnId: null,
turnState: null,
turnKind: 'normal',
replyAttempt: 0,
replyRetryTurnId: null,
finalText: '',
waitingUser: null,
lastVerification: null,
createdAt: isoNow(now),
updatedAt: isoNow(now),
metadata: clone(input.metadata || {}),
};
tasks.set(taskId, task);
if (task.threadId) sessions.set(sessionKey, { sessionKey, threadId: task.threadId, updatedAt: task.updatedAt });
return clone(task);
}
function getTask(taskId) {
return clone(tasks.get(cleanText(taskId, 256)) || null);
}
function listTasks() {
return Array.from(tasks.values(), clone);
}
function threadConfig(input = {}) {
return buildGiteaThreadConfig(input);
}
function attachThread(taskId, threadId) {
const task = tasks.get(requiredText(taskId, 'taskId', 256));
const nextThreadId = requiredText(threadId, 'threadId', 256);
if (task.threadId && task.threadId !== nextThreadId) throw new Error('同一工作流任务不能切换 threadId。');
task.threadId = nextThreadId;
sessions.set(task.sessionKey, { sessionKey: task.sessionKey, threadId: nextThreadId, updatedAt: isoNow(now) });
return emit(task, 'thread_attached');
}
function handleTurnStarted(input = {}) {
const task = tasks.get(requiredText(input.taskId, 'taskId', 256));
const turnId = requiredText(input.turnId || extractTurnId(input), 'turnId', 256);
const threadId = cleanText(input.threadId || extractThreadId(input), 256) || task.threadId;
if (threadId) attachThread(task.taskId, threadId);
if (task.turnId && task.turnId !== turnId && task.state === WORKFLOW_STATES.RUNNING) {
throw new Error('任务已有运行中的 turnId。');
}
task.turnId = turnId;
task.turnKind = input.kind === 'reply_retry' ? 'reply_retry' : 'normal';
task.turnState = TURN_STATES.STARTED;
task.state = WORKFLOW_STATES.RUNNING;
if (task.turnKind === 'reply_retry') task.replyRetryTurnId = turnId;
return emit(task, 'turn_started', { turnId, threadId: task.threadId });
}
function handleTurnEvent(input = {}) {
const task = tasks.get(requiredText(input.taskId, 'taskId', 256));
const turnId = extractTurnId(input.notification || input) || cleanText(input.turnId, 256) || task.turnId;
if (turnId && task.turnId && turnId !== task.turnId) return { ok: false, code: 'turn_id_mismatch', task: clone(task) };
if (turnId && !task.turnId) task.turnId = turnId;
if (isWaitingUserEvent(input.notification || input)) {
task.turnState = TURN_STATES.WAITING_USER;
task.state = WORKFLOW_STATES.WAITING_USER;
task.waitingUser = clone(input.notification?.params || input.params || input);
return emit(task, 'waiting_user', { turnId: task.turnId });
}
if (String(input.method || input.notification?.method || '').toLowerCase() === 'turn/started') {
task.turnState = TURN_STATES.STARTED;
task.state = WORKFLOW_STATES.RUNNING;
}
return emit(task, 'turn_event', { turnId: task.turnId, method: input.method || input.notification?.method || null });
}
async function verifyReply(task, turnId, source) {
const markerText = buildWorkflowMarker({ taskId: task.taskId, turnId, resourceKey: task.resourceKey, kind: 'final' });
const expected = { taskId: task.taskId, turnId, resourceKey: task.resourceKey };
let result;
try {
if (typeof options.verifyReply === 'function') {
result = await options.verifyReply({ task: clone(task), expected, source, marker: markerText });
} else if (typeof options.listComments === 'function') {
result = await options.listComments({ task: clone(task), expected, source });
} else {
result = { status: 'unknown', error: '未配置 Gitea 回执查询器。' };
}
} catch (error) {
result = { status: 'unknown', error: error?.message || String(error || '') };
}
const normalized = normalizeVerification(result, expected, botIdentity);
task.lastVerification = { ...normalized, source, turnId, checkedAt: isoNow(now) };
return normalized;
}
async function handleTurnCompleted(input = {}) {
const task = tasks.get(requiredText(input.taskId, 'taskId', 256));
const turnId = cleanText(input.turnId || extractTurnId(input), 256) || task.turnId;
if (!turnId || (task.turnId && turnId !== task.turnId)) return { ok: false, code: 'turn_id_mismatch', task: clone(task) };
task.turnId = turnId;
if (input.waitingUser || isWaitingUserEvent(input.notification || input)) {
task.turnState = TURN_STATES.WAITING_USER;
task.state = WORKFLOW_STATES.WAITING_USER;
task.waitingUser = clone(input.waitingUser || input.notification?.params || input.params || {});
return emit(task, 'waiting_user', { turnId });
}
const status = String(input.status || input.stopReason || 'completed').toLowerCase();
if (['failed', 'error'].includes(status)) {
task.turnState = TURN_STATES.FAILED;
task.state = WORKFLOW_STATES.FAILED;
return emit(task, 'turn_failed', { turnId });
}
if (['interrupted', 'cancelled', 'canceled', 'aborted'].includes(status)) {
task.turnState = TURN_STATES.INTERRUPTED;
task.state = WORKFLOW_STATES.FAILED;
return emit(task, 'turn_interrupted', { turnId });
}
task.turnState = TURN_STATES.COMPLETED;
task.finalText = String(input.finalText || task.finalText || '').trim();
task.state = WORKFLOW_STATES.VERIFYING_REPLY;
const first = await verifyReply(task, turnId, 'mcp');
if (first.status === 'confirmed') {
task.state = WORKFLOW_STATES.SUCCEEDED;
return emit(task, 'reply_confirmed', { turnId, source: 'mcp', commentId: first.commentId || first.comment?.id || null });
}
if (first.status === 'unknown') {
// 查询异常代表外部状态未知;不能据此再次执行修改、补发或 REST 覆盖。
task.state = WORKFLOW_STATES.FAILED_REPLY;
return emit(task, 'reply_verification_unknown', { turnId, verification: first });
}
if (task.replyAttempt < replyRetryLimit && typeof options.startTurn === 'function') {
task.replyAttempt += 1;
task.state = WORKFLOW_STATES.RETRYING_REPLY;
const marker = {
turnId,
text: buildWorkflowMarker({ taskId: task.taskId, turnId, resourceKey: task.resourceKey, kind: 'final' }),
};
const prompt = buildReplyRetryPrompt(task, task.finalText, marker);
try {
const retry = await options.startTurn({
task: clone(task),
threadId: task.threadId,
kind: 'reply_retry',
hidden: true,
prompt,
marker: marker.text,
});
const retryTurnId = cleanText(retry?.turnId || extractTurnId(retry), 256);
if (!retryTurnId) throw new Error('补触发未返回 turnId。');
task.turnId = retryTurnId;
task.replyRetryTurnId = retryTurnId;
task.turnKind = 'reply_retry';
task.turnState = TURN_STATES.STARTED;
task.state = WORKFLOW_STATES.RUNNING;
return emit(task, 'reply_retry_started', { turnId: retryTurnId, previousTurnId: turnId, threadId: task.threadId });
} catch (error) {
task.lastVerification = { ...task.lastVerification, retryError: error?.message || String(error || '') };
}
}
// 补触发失败或补触发后的 turn 完成会落到这里REST 是唯一最后出口。
const markerText = buildWorkflowMarker({ taskId: task.taskId, turnId: task.turnId || turnId, resourceKey: task.resourceKey, kind: 'final' });
let restResult = null;
if (typeof options.sendRestReply === 'function') {
try {
restResult = await options.sendRestReply({
task: clone(task),
body: `${task.finalText}${task.finalText ? '\n\n' : ''}${markerText}`,
marker: markerText,
source: 'rest_fallback',
});
} catch (error) {
restResult = { status: 'unknown', error: error?.message || String(error || '') };
}
}
const fallbackTurnId = task.turnId || turnId;
const fallbackVerification = await verifyReply(task, fallbackTurnId, 'rest_fallback');
if (fallbackVerification.status === 'confirmed' || restResult?.confirmed === true) {
task.state = WORKFLOW_STATES.SUCCEEDED_WITH_REST_FALLBACK;
return emit(task, 'reply_confirmed', { turnId: fallbackTurnId, source: 'rest_fallback', commentId: fallbackVerification.commentId || fallbackVerification.comment?.id || restResult?.commentId || null });
}
task.state = WORKFLOW_STATES.FAILED_REPLY;
return emit(task, 'reply_failed', { turnId: fallbackTurnId, source: 'rest_fallback', verification: fallbackVerification });
}
function queueWaitingUserComment(input = {}) {
const parent = tasks.get(requiredText(input.taskId, 'taskId', 256));
const comment = normalizeComment(input.comment || {});
if (isBotComment(input.comment || {}, botIdentity) || parseWorkflowMarker(comment.body)) {
return { ok: true, ignored: true, reason: 'bot_self_comment' };
}
const taskId = requiredText(input.nextTaskId || input.comment?.taskId || `${parent.taskId}:continuation:${comment.id || Date.now()}`, 'nextTaskId', 256);
const next = createTask({
taskId,
sessionKey: parent.sessionKey,
resourceKey: parent.resourceKey,
threadId: parent.threadId,
parentTaskId: parent.taskId,
metadata: { continuation: true, commentId: comment.id },
});
next.prompt = buildWaitingUserPrompt(comment);
const stored = tasks.get(taskId);
stored.prompt = next.prompt;
stored.waitingUser = null;
return emit(stored, 'waiting_user_comment_queued', { parentTaskId: parent.taskId, threadId: parent.threadId });
}
return {
createTask,
getTask,
listTasks,
threadConfig,
attachThread,
handleTurnStarted,
handleTurnEvent,
handleTurnCompleted,
queueWaitingUserComment,
buildWorkflowMarker,
parseWorkflowMarker,
isBotComment: (comment) => isBotComment(comment, botIdentity),
matchWorkflowReply: (comment, expected) => matchWorkflowReply(comment, expected, botIdentity),
states: WORKFLOW_STATES,
turnStates: TURN_STATES,
};
}
module.exports = {
DEFAULT_GITEA_MCP_COMMAND,
WORKFLOW_MARKER_NAME,
WORKFLOW_MARKER_VERSION,
WORKFLOW_STATES,
TURN_STATES,
buildGiteaMcpServerConfig,
buildGiteaMcpConfig,
buildGiteaThreadConfig,
mergeGiteaThreadConfig,
extractTurnId,
extractThreadId,
buildWorkflowMarker,
parseWorkflowMarker,
isBotComment,
normalizeComment,
matchWorkflowReply,
findWorkflowReply,
isWaitingUserEvent,
buildWaitingUserPrompt,
createGiteaWorkflowCodex,
encodeMarker,
buildWorkflowPrompt,
classifyReplyComments,
};

View File

@@ -0,0 +1,318 @@
'use strict';
/**
* Gitea Workflow 领域模型。
*
* 该文件只负责无副作用的数据规范化、标识生成和状态机校验,便于
* Webhook、调度器、Codex App 适配器以及管理 API 共享同一套契约。
*/
const crypto = require('crypto');
const TASK_STATES = Object.freeze({
RECEIVED: 'received',
QUEUED: 'queued',
PREPARING: 'preparing',
RUNNING: 'running',
WAITING_USER: 'waiting_user',
VERIFYING_REPLY: 'verifying_reply',
RETRY_WAIT: 'retry_wait',
BLOCKED_WORKSPACE: 'blocked_workspace',
ABORTING: 'aborting',
SUCCEEDED: 'succeeded',
SUCCEEDED_WITH_REST_FALLBACK: 'succeeded_with_rest_fallback',
FAILED: 'failed',
FAILED_REPLY: 'failed_reply',
CANCELLED: 'cancelled',
ABORTED: 'aborted',
IGNORED: 'ignored',
DUPLICATE: 'duplicate',
REJECTED: 'rejected',
});
const TERMINAL_STATES = new Set([
TASK_STATES.SUCCEEDED,
TASK_STATES.SUCCEEDED_WITH_REST_FALLBACK,
TASK_STATES.FAILED,
TASK_STATES.FAILED_REPLY,
TASK_STATES.CANCELLED,
TASK_STATES.ABORTED,
TASK_STATES.IGNORED,
TASK_STATES.DUPLICATE,
TASK_STATES.REJECTED,
]);
const RUNNING_STATES = new Set([
TASK_STATES.PREPARING,
TASK_STATES.RUNNING,
TASK_STATES.ABORTING,
]);
const TASK_TRANSITIONS = Object.freeze({
received: new Set(['queued', 'ignored', 'duplicate', 'rejected']),
queued: new Set(['preparing', 'cancelled']),
preparing: new Set(['running', 'blocked_workspace', 'retry_wait', 'failed']),
running: new Set(['waiting_user', 'verifying_reply', 'retry_wait', 'blocked_workspace', 'aborting', 'failed']),
waiting_user: new Set(['queued', 'cancelled', 'failed']),
verifying_reply: new Set(['succeeded', 'succeeded_with_rest_fallback', 'retry_wait', 'failed_reply']),
retry_wait: new Set(['preparing', 'running', 'verifying_reply', 'failed']),
blocked_workspace: new Set(['queued', 'cancelled']),
aborting: new Set(['aborted', 'failed']),
});
const TURN_STATES = Object.freeze({
CREATED: 'created',
STARTING: 'starting',
RUNNING: 'running',
WAITING_USER: 'waiting_user',
COMPLETED: 'completed',
FAILED: 'failed',
ABORTING: 'aborting',
ABORTED: 'aborted',
});
function isObject(value) {
return Boolean(value) && typeof value === 'object' && !Array.isArray(value);
}
function clone(value) {
if (value === undefined) return undefined;
return JSON.parse(JSON.stringify(value));
}
function text(value, max = 1000) {
return typeof value === 'string' ? value.trim().slice(0, max) : '';
}
function integer(value, fallback = 0) {
return Number.isSafeInteger(value) ? value : (Number.isSafeInteger(Number(value)) ? Number(value) : fallback);
}
function iso(now = Date.now) {
const value = typeof now === 'function' ? now() : now;
const date = value instanceof Date ? value : new Date(value);
return Number.isFinite(date.getTime()) ? date.toISOString() : new Date().toISOString();
}
function id(prefix) {
return `${prefix}-${crypto.randomUUID()}`;
}
function digest(value, length = 24) {
return crypto.createHash('sha256').update(String(value)).digest('hex').slice(0, length);
}
function sessionKeyFor({ instanceId = 'default', owner, repo, kind, number } = {}) {
const normalizedKind = kind === 'pull_request' || kind === 'pr' ? 'pull_request' : 'issue';
const ownerName = text(owner, 160);
const repoName = text(repo, 160);
const issueNumber = integer(number, 0);
if (!ownerName || !repoName || issueNumber <= 0) throw new TypeError('sessionKey 缺少 owner/repo/kind/number');
return `${text(instanceId, 80) || 'default'}:${ownerName}/${repoName}:${normalizedKind}:${issueNumber}`;
}
function resourceKeyFor(input = {}) {
// 资源身份必须包含编号,才能在回帖确认时区分同一仓库的不同 Issue/PR。
return sessionKeyFor(input);
}
function repoKeyFor({ instanceId = 'default', owner, repo, name } = {}) {
const repoName = text(repo || name, 160);
const ownerName = text(owner, 160);
if (!ownerName || !repoName) throw new TypeError('repoKey 缺少 owner/repo');
return `${text(instanceId, 80) || 'default'}:${ownerName}/${repoName}`;
}
function normalizeConfig(input = {}, env = process.env) {
const gitea = isObject(input.gitea) ? input.gitea : {};
const codex = isObject(input.codex) ? input.codex : {};
const workspaceRoot = text(input.workspaceRoot || env.CC_WEB_GITEA_WORKSPACE_ROOT, 1000)
|| require('path').join(process.cwd(), 'gitea-workspaces');
return Object.freeze({
instanceId: text(input.instanceId || env.CC_WEB_GITEA_INSTANCE_ID, 80) || 'default',
botLogin: text(input.botLogin || env.CC_WEB_GITEA_BOT_LOGIN, 120) || 'ccweb-bot',
botId: input.botId ?? env.CC_WEB_GITEA_BOT_ID ?? null,
workspaceRoot,
gitea: Object.freeze({
host: text(gitea.host || input.giteaHost || env.CC_WEB_GITEA_HOST, 500),
defaultBranchOverride: text(gitea.defaultBranchOverride || input.defaultBranchOverride || env.CC_WEB_GITEA_DEFAULT_BRANCH, 200) || '',
token: String(gitea.token ?? input.giteaToken ?? env.CC_WEB_GITEA_TOKEN ?? ''),
webhookSecret: String(gitea.webhookSecret ?? input.webhookSecret ?? env.CC_WEB_GITEA_WEBHOOK_SECRET ?? ''),
}),
codex: Object.freeze({
command: text(codex.command || env.CC_WEB_CODEX_APP_COMMAND, 200) || 'codexapp',
mode: 'yolo',
model: text(codex.model || env.CC_WEB_CODEX_MODEL, 120) || null,
reasoningEffort: text(codex.reasoningEffort || env.CC_WEB_CODEX_REASONING_EFFORT, 80) || null,
developerInstructions: text(codex.developerInstructions, 4000) || null,
}),
});
}
function publicConfig(config) {
const value = clone(config || {});
if (value.gitea) {
value.gitea.token = value.gitea.token ? `sha256:${digest(value.gitea.token, 16)}` : null;
value.gitea.webhookSecret = value.gitea.webhookSecret ? `sha256:${digest(value.gitea.webhookSecret, 16)}` : null;
}
return value;
}
function createRepositoryRecord(input = {}, options = {}) {
const now = iso(options.now);
const owner = text(input.owner, 160);
const name = text(input.name || input.repo, 160);
if (!owner || !name) throw new TypeError('仓库记录缺少 owner/name');
return {
key: input.key || repoKeyFor({ instanceId: input.instanceId, owner, repo: name }),
instanceId: text(input.instanceId, 80) || 'default', owner, name,
fullName: `${owner}/${name}`,
cloneUrl: text(input.cloneUrl || input.clone_url, 600) || null,
defaultBranch: text(input.defaultBranch || input.default_branch, 200) || null,
workspacePath: text(input.workspacePath, 1200) || null,
lastFetchAt: input.lastFetchAt || null,
status: text(input.status, 40) || 'active',
enabled: input.enabled !== false && input.status !== 'disabled',
dirty: input.dirty === true,
dirtyReason: text(input.dirtyReason, 1000) || null,
dirtySessionKey: text(input.dirtySessionKey, 300) || null,
createdAt: input.createdAt || now,
updatedAt: input.updatedAt || now,
version: Math.max(0, integer(input.version, 0)),
activeTaskId: text(input.activeTaskId, 160) || null,
};
}
function createSessionRecord(input = {}, options = {}) {
const now = iso(options.now);
const sessionKey = input.sessionKey || sessionKeyFor(input);
return {
sessionKey,
instanceId: text(input.instanceId, 80) || 'default',
owner: text(input.owner, 160), repo: text(input.repo || input.name, 160),
kind: input.kind === 'pull_request' || input.kind === 'pr' ? 'pull_request' : 'issue',
number: integer(input.number, 0),
resourceKey: input.resourceKey || resourceKeyFor(input),
threadId: text(input.threadId, 240) || null,
ccwebSessionId: text(input.ccwebSessionId, 240) || null,
status: text(input.status, 40) || 'idle',
createdAt: input.createdAt || now, updatedAt: input.updatedAt || now,
lastTurnId: text(input.lastTurnId, 240) || null,
version: Math.max(0, integer(input.version, 0)),
};
}
function createTaskRecord(input = {}, options = {}) {
const now = iso(options.now);
const taskId = text(input.taskId || input.id, 180) || id('gitea-task');
const state = text(input.state || input.status, 40) || TASK_STATES.RECEIVED;
if (!Object.values(TASK_STATES).includes(state)) throw new TypeError(`未知任务状态: ${state}`);
return {
taskId,
deliveryKey: text(input.deliveryKey, 240) || null,
deliveryId: text(input.deliveryId, 180) || null,
repoKey: text(input.repoKey, 240) || null,
resourceKey: text(input.resourceKey, 300) || null,
sessionKey: text(input.sessionKey, 300) || null,
commentId: input.commentId ?? null,
actor: text(input.actor || input.author, 180) || null,
instruction: text(input.instruction || input.prompt, 8000),
state,
stateVersion: Math.max(0, integer(input.stateVersion, 0)),
attempt: Math.max(0, integer(input.attempt, 0)),
replyAttempt: Math.max(0, integer(input.replyAttempt, 0)),
threadId: text(input.threadId, 240) || null,
turnId: text(input.turnId, 240) || null,
turnState: text(input.turnState, 40) || null,
turnStartedAt: input.turnStartedAt || null,
turnUpdatedAt: input.turnUpdatedAt || null,
createdAt: input.createdAt || now, updatedAt: input.updatedAt || now,
nextRetryAt: input.nextRetryAt || null,
errorCode: text(input.errorCode, 160) || null,
errorMessage: text(input.errorMessage || input.error, 1200) || null,
metadata: isObject(input.metadata) ? clone(input.metadata) : {},
};
}
function createTurnRecord(input = {}, options = {}) {
const now = iso(options.now);
return {
turnId: text(input.turnId, 240) || id('gitea-turn'),
taskId: text(input.taskId, 180) || null,
sessionKey: text(input.sessionKey, 300) || null,
threadId: text(input.threadId, 240) || null,
state: text(input.state, 40) || TURN_STATES.CREATED,
startedAt: input.startedAt || null,
finishedAt: input.finishedAt || null,
lastEventSeq: Math.max(0, integer(input.lastEventSeq, 0)),
retryCount: Math.max(0, integer(input.retryCount, 0)),
createdAt: input.createdAt || now,
updatedAt: input.updatedAt || now,
};
}
function canTransition(from, to) {
if (from === to) return true;
return Boolean(TASK_TRANSITIONS[from]?.has(to));
}
function transitionTask(task, nextState, options = {}) {
const current = task?.state || task?.status;
if (!canTransition(current, nextState)) {
const error = new Error(`任务状态不可从 ${current} 转为 ${nextState}`);
error.code = 'invalid_task_transition';
error.fromState = current;
error.toState = nextState;
throw error;
}
const next = { ...clone(task), state: nextState, status: nextState,
stateVersion: integer(task.stateVersion, 0) + (current === nextState ? 0 : 1),
updatedAt: iso(options.now) };
if (options.turnId !== undefined) next.turnId = options.turnId;
if (options.errorCode !== undefined) next.errorCode = options.errorCode;
if (options.errorMessage !== undefined) next.errorMessage = options.errorMessage;
if (options.nextRetryAt !== undefined) next.nextRetryAt = options.nextRetryAt;
return next;
}
function recoverTasks(tasks, options = {}) {
const now = iso(options.now);
return (Array.isArray(tasks) ? tasks : []).map((task) => {
if (task.state === TASK_STATES.RUNNING || task.state === TASK_STATES.PREPARING) {
const retries = integer(task.attempt, 0);
if (retries < (options.maxRestartRetries ?? 1)) {
const recovered = transitionTask(task, TASK_STATES.RETRY_WAIT, { now: options.now, errorCode: 'interrupted_by_restart', errorMessage: '进程重启导致任务中断。', nextRetryAt: now });
recovered.attempt = retries + 1;
return recovered;
}
return transitionTask(task, TASK_STATES.FAILED, { now: options.now, errorCode: 'restart_retry_exhausted', errorMessage: '重启恢复重试次数已耗尽。' });
}
if (task.state === TASK_STATES.ABORTING) {
return transitionTask(task, TASK_STATES.FAILED, { now: options.now, errorCode: 'aborting_interrupted_by_restart', errorMessage: '中止确认期间进程重启。' });
}
return clone(task);
});
}
module.exports = {
TASK_STATES,
TERMINAL_STATES,
RUNNING_STATES,
TASK_TRANSITIONS,
TURN_STATES,
canTransition,
clone,
createRepositoryRecord,
createSessionRecord,
createTaskRecord,
createTurnRecord,
digest,
iso,
normalizeConfig,
publicConfig,
recoverTasks,
repoKeyFor,
resourceKeyFor,
sessionKeyFor,
transitionTask,
};

View File

@@ -0,0 +1,538 @@
'use strict';
/**
* Gitea Workflow 管理面适配器。
*
* 该模块不实现 Webhook、调度器或 Codex turn只把核心 workflow service
* 暴露为稳定的查询/控制契约,并在核心服务尚未挂接时提供可恢复的只读状态。
*/
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const TERMINAL_STATES = new Set([
'succeeded', 'succeeded_with_rest_fallback', 'failed', 'failed_reply',
'cancelled', 'aborted', 'ignored', 'duplicate', 'rejected',
]);
const RUNNING_STATES = new Set(['preparing', 'running', 'aborting']);
function nowIso() {
return new Date().toISOString();
}
function clone(value) {
return value == null ? value : JSON.parse(JSON.stringify(value));
}
function safeText(value, max = 2000) {
return String(value == null ? '' : value).slice(0, max);
}
function requestId(value) {
const text = safeText(value, 120).trim();
return text || crypto.randomUUID();
}
function stripGlobalConcurrency(settings = {}) {
const next = { ...settings };
delete next.maxConcurrency;
return next;
}
function createEmptyState() {
return {
version: 1,
control: {
globalPaused: false,
pauseReason: null,
updatedBy: 'system',
updatedAt: nowIso(),
version: 0,
},
repositories: [],
tasks: [],
audits: [],
logs: [],
controlRequests: {},
settings: {
host: '', botLogin: 'ccweb-bot', workspaceRoot: '', defaultBranch: '',
webhookSecretConfigured: false, botTokenConfigured: false,
updatedAt: nowIso(), updatedBy: 'system',
},
updatedAt: nowIso(),
};
}
function readState(filePath) {
if (!filePath) return createEmptyState();
try {
if (!fs.existsSync(filePath)) return createEmptyState();
const stat = fs.statSync(filePath);
if (stat.size > 8 * 1024 * 1024) throw new Error('workflow 管理状态文件超过 8MB 上限');
const parsed = JSON.parse(fs.readFileSync(filePath, 'utf8'));
return {
...createEmptyState(),
...parsed,
control: { ...createEmptyState().control, ...(parsed.control || {}) },
repositories: Array.isArray(parsed.repositories) ? parsed.repositories : [],
tasks: Array.isArray(parsed.tasks) ? parsed.tasks : [],
audits: Array.isArray(parsed.audits) ? parsed.audits : [],
logs: Array.isArray(parsed.logs) ? parsed.logs : [],
controlRequests: parsed.controlRequests && typeof parsed.controlRequests === 'object' ? parsed.controlRequests : {},
settings: stripGlobalConcurrency({ ...createEmptyState().settings, ...(parsed.settings || {}) }),
};
} catch {
return createEmptyState();
}
}
function writeState(filePath, state) {
if (!filePath) return;
fs.mkdirSync(path.dirname(filePath), { recursive: true });
const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
fs.writeFileSync(tempPath, JSON.stringify(state, null, 2), { mode: 0o600 });
fs.renameSync(tempPath, filePath);
}
function normalizeRepository(repository = {}) {
const status = String(repository.status || (repository.enabled === false ? 'disabled' : 'active'));
return {
repoKey: safeText(repository.repoKey || repository.key || '', 240),
instanceId: safeText(repository.instanceId || 'default', 80),
owner: safeText(repository.owner || '', 120),
name: safeText(repository.name || repository.repo || '', 160),
cloneUrl: safeText(repository.cloneUrl || '', 500),
workspacePath: safeText(repository.workspacePath || repository.path || '', 1000),
defaultBranch: safeText(repository.defaultBranch || repository.default_branch || '', 160),
status,
enabled: status !== 'disabled',
registeredAt: repository.registeredAt || repository.createdAt || null,
lastFetchAt: repository.lastFetchAt || null,
dirty: !!(repository.dirty || repository.workspaceDirty || status === 'blocked'),
dirtyReason: safeText(repository.dirtyReason || repository.blockedReason || repository.workspaceReason || '', 1000) || null,
dirtySessionKey: safeText(repository.dirtySessionKey || '', 300) || null,
activeTaskId: safeText(repository.activeTaskId || '', 120) || null,
queuedCount: Number(repository.queuedCount || 0),
version: Number(repository.version || 0),
};
}
function normalizeTask(task = {}) {
const state = safeText(task.state || task.status || 'queued', 64) || 'queued';
return {
taskId: safeText(task.taskId || task.id || '', 120),
repoKey: safeText(task.repoKey || '', 240),
resourceKey: safeText(task.resourceKey || '', 300),
sessionKey: safeText(task.sessionKey || '', 300),
commentId: task.commentId ?? null,
author: safeText(task.author || task.actor || '', 160),
instruction: safeText(task.instruction || task.prompt || '', 4000),
state,
stateLabel: safeText(task.stateLabel || '', 120) || null,
attempt: Number(task.attempt || 0),
replyAttempt: Number(task.replyAttempt || 0),
threadId: safeText(task.threadId || '', 240) || null,
turnId: safeText(task.turnId || '', 240) || null,
turnState: safeText(task.turnState || task.turn?.status || '', 80) || null,
turnStartedAt: task.turnStartedAt || task.turn?.startedAt || null,
turnUpdatedAt: task.turnUpdatedAt || task.turn?.updatedAt || null,
logs: Array.isArray(task.logs) ? task.logs.slice(-100).map((entry) => ({
timestamp: entry.timestamp || entry.ts || null,
level: safeText(entry.level || 'info', 20),
message: safeText(entry.message || entry.event || '', 1000),
})) : [],
dirtyReason: safeText(task.dirtyReason || task.workspaceReason || (state === 'blocked_workspace' ? task.errorMessage || task.error : ''), 1000) || null,
errorCode: safeText(task.errorCode || '', 120) || null,
errorMessage: safeText(task.errorMessage || task.error || '', 1000) || null,
createdAt: task.createdAt || null,
updatedAt: task.updatedAt || null,
nextRetryAt: task.nextRetryAt || null,
terminal: TERMINAL_STATES.has(state),
};
}
function normalizeAudit(entry = {}) {
return {
eventId: safeText(entry.eventId || entry.id || '', 120),
taskId: safeText(entry.taskId || '', 120) || null,
repoKey: safeText(entry.repoKey || '', 240) || null,
action: safeText(entry.action || entry.event || '', 160),
actor: safeText(entry.actor || entry.operator || 'system', 160),
reason: safeText(entry.reason || '', 1000) || null,
fromState: safeText(entry.fromState || '', 64) || null,
toState: safeText(entry.toState || '', 64) || null,
deliveryId: safeText(entry.deliveryId || '', 160) || null,
turnId: safeText(entry.turnId || '', 240) || null,
commentId: entry.commentId ?? null,
errorCode: safeText(entry.errorCode || '', 120) || null,
timestamp: entry.timestamp || entry.createdAt || entry.ts || null,
metadata: entry.metadata && typeof entry.metadata === 'object' ? clone(entry.metadata) : {},
};
}
function normalizeLog(entry = {}) {
return {
timestamp: entry.timestamp || entry.ts || null,
level: safeText(entry.level || 'info', 20),
taskId: safeText(entry.taskId || '', 120) || null,
repoKey: safeText(entry.repoKey || '', 240) || null,
event: safeText(entry.event || entry.action || '', 160),
message: safeText(entry.message || entry.error || '', 1600),
};
}
function pickService(service, names, ...args) {
if (!service) return undefined;
for (const name of names) {
if (typeof service[name] !== 'function') continue;
return service[name](...args);
}
return undefined;
}
function collectionItems(value) {
if (Array.isArray(value)) return value;
if (Array.isArray(value?.items)) return value.items;
return [];
}
function createGiteaWorkflowManagement(options = {}) {
const statePath = options.statePath || null;
let service = options.workflowService || null;
const clock = options.now || nowIso;
let state = readState(statePath);
const secretsPath = options.secretsPath || null;
const onConfigUpdate = typeof options.onConfigUpdate === 'function' ? options.onConfigUpdate : null;
function readSecrets() {
if (!secretsPath) return {};
try { return JSON.parse(fs.readFileSync(secretsPath, 'utf8')); } catch { return {}; }
}
function writeSecrets(next) {
if (!secretsPath) return;
fs.mkdirSync(path.dirname(secretsPath), { recursive: true });
const temp = `${secretsPath}.${process.pid}.${Date.now()}.tmp`;
fs.writeFileSync(temp, JSON.stringify(next, null, 2), { mode: 0o600 });
fs.renameSync(temp, secretsPath);
}
function configuration() {
const settings = stripGlobalConcurrency({ ...state.settings });
const secrets = readSecrets();
settings.webhookSecretConfigured = Boolean(secrets.webhookSecret || settings.webhookSecretConfigured);
settings.botTokenConfigured = Boolean(secrets.botToken || settings.botTokenConfigured);
return { settings, secrets: { webhookSecretConfigured: settings.webhookSecretConfigured, botTokenConfigured: settings.botTokenConfigured } };
}
function runtimeConfiguration() {
const publicConfig = configuration();
const secrets = readSecrets();
return { settings: stripGlobalConcurrency(publicConfig.settings), secrets: { webhookSecret: secrets.webhookSecret || '', botToken: secrets.botToken || '' } };
}
function updateConfiguration(input = {}) {
const actor = safeText(input.actor || 'admin', 160);
const reason = safeText(input.reason || '', 1000);
if (!reason) throw Object.assign(new Error('配置变更必须填写操作原因。'), { statusCode: 400, code: 'reason_required' });
const current = runtimeConfiguration();
const requestedWorkspace = safeText(input.workspaceRoot ?? state.settings.workspaceRoot, 1200);
const requestedHost = safeText(input.host ?? state.settings.host, 500);
if (requestedHost) {
let parsedHost;
try { parsedHost = new URL(requestedHost); } catch {
throw Object.assign(new Error('Gitea 地址必须是完整的 http/https URL。'), { statusCode: 400, code: 'invalid_gitea_host' });
}
if (!['http:', 'https:'].includes(parsedHost.protocol) || parsedHost.username || parsedHost.password) {
throw Object.assign(new Error('Gitea 地址只允许 http/https且不能携带凭据。'), { statusCode: 400, code: 'invalid_gitea_host' });
}
}
if (requestedWorkspace && (!path.isAbsolute(requestedWorkspace) || path.resolve(requestedWorkspace) === path.parse(requestedWorkspace).root)) {
throw Object.assign(new Error('工作区根目录必须是非根的绝对路径。'), { statusCode: 400, code: 'invalid_workspace_root' });
}
const nextSettings = {
...state.settings,
host: requestedHost,
botLogin: safeText(input.botLogin ?? state.settings.botLogin, 120) || 'ccweb-bot',
workspaceRoot: requestedWorkspace,
defaultBranch: safeText(input.defaultBranch ?? state.settings.defaultBranch, 200),
updatedAt: clock(), updatedBy: actor,
};
const secrets = { ...current.secrets };
if (Object.prototype.hasOwnProperty.call(input, 'webhookSecret') && String(input.webhookSecret || '').trim()) secrets.webhookSecret = String(input.webhookSecret).trim();
if (Object.prototype.hasOwnProperty.call(input, 'botToken') && String(input.botToken || '').trim()) secrets.botToken = String(input.botToken).trim();
nextSettings.webhookSecretConfigured = Boolean(secrets.webhookSecret);
nextSettings.botTokenConfigured = Boolean(secrets.botToken);
state.settings = stripGlobalConcurrency(nextSettings);
writeSecrets(secrets);
appendAudit({ action: 'admin.configuration_updated', actor, reason, metadata: { changed: Object.keys(input).filter((key) => !['webhookSecret', 'botToken', 'maxConcurrency'].includes(key)) } });
if (onConfigUpdate) onConfigUpdate({ settings: clone(state.settings), secrets: clone(secrets) });
persist();
return { settings: clone(state.settings), secrets: { webhookSecretConfigured: nextSettings.webhookSecretConfigured, botTokenConfigured: nextSettings.botTokenConfigured } };
}
function persist() {
state.updatedAt = clock();
writeState(statePath, state);
}
function appendAudit({ action, actor = 'system', reason = '', taskId = null, repoKey = null, fromState = null, toState = null, metadata = {} }) {
const item = normalizeAudit({
eventId: crypto.randomUUID(), taskId, repoKey, action, actor, reason,
fromState, toState, metadata, timestamp: clock(),
});
state.audits.push(item);
state.audits = state.audits.slice(-2000);
persist();
return item;
}
function sourceOverview() {
const value = pickService(service, ['getManagementOverview', 'getOverview', 'overview']);
return value && typeof value === 'object' ? value : null;
}
function snapshot() {
const overview = sourceOverview() || {};
const repositories = pickService(service, ['listRepositories', 'getRepositories']) || overview.repositories || state.repositories;
const tasks = pickService(service, ['listTasks', 'getTasks'], {}) || overview.tasks || state.tasks;
const auditSource = pickService(service, ['listAudit', 'listAudits', 'getAudit']);
const logSource = pickService(service, ['listLogs', 'getLogs']);
const audits = auditSource
? [...state.audits, ...collectionItems(auditSource)]
: (overview.audits || state.audits);
const logs = logSource
? [...state.logs, ...collectionItems(logSource)]
: (overview.logs || state.logs);
const normalizedTasks = collectionItems(tasks).map(normalizeTask);
const normalizedRepos = collectionItems(repositories).map(normalizeRepository);
const tasksByRepo = new Map();
for (const task of normalizedTasks) {
if (!task.repoKey) continue;
const list = tasksByRepo.get(task.repoKey) || [];
list.push(task);
tasksByRepo.set(task.repoKey, list);
}
for (const repo of normalizedRepos) {
const repoTasks = tasksByRepo.get(repo.repoKey) || [];
repo.queuedCount = repoTasks.filter((task) => task.state === 'queued' || task.state === 'retry_wait').length;
repo.activeTaskId = repo.activeTaskId || repoTasks.find((task) => RUNNING_STATES.has(task.state))?.taskId || null;
const blockedTask = repoTasks.find((task) => task.state === 'blocked_workspace');
if (blockedTask) {
repo.dirty = true;
repo.dirtyReason = repo.dirtyReason || blockedTask.dirtyReason || blockedTask.errorMessage || '工作区阻塞';
repo.status = 'blocked';
}
}
const runningCount = normalizedTasks.filter((task) => RUNNING_STATES.has(task.state)).length;
const queuedCount = normalizedTasks.filter((task) => task.state === 'queued' || task.state === 'retry_wait').length;
const control = {
...state.control,
...(overview.control || {}),
globalPaused: overview.globalPaused ?? overview.control?.globalPaused ?? state.control.globalPaused,
};
const taskLogs = normalizedTasks.flatMap((task) => task.logs.map((entry) => ({ ...entry, taskId: task.taskId, repoKey: task.repoKey })));
const normalizedLogs = (Array.isArray(logs) ? logs : []).map(normalizeLog);
const auditLogs = collectionItems(audits).map((entry) => ({
timestamp: entry.timestamp || entry.createdAt || entry.ts || null,
level: entry.errorCode ? 'warn' : 'info',
taskId: entry.taskId || null,
repoKey: entry.repoKey || null,
event: entry.action || entry.event || 'workflow',
message: [entry.action || entry.event || 'workflow', entry.errorCode, entry.toState].filter(Boolean).join(' · '),
})).map(normalizeLog);
return {
ok: true,
available: !!service,
generatedAt: clock(),
control,
summary: {
running: Number(overview.summary?.running ?? runningCount),
queued: Number(overview.summary?.queued ?? queuedCount),
repositories: normalizedRepos.length,
blocked: normalizedRepos.filter((repo) => repo.dirty || repo.status === 'blocked').length,
},
repositories: normalizedRepos,
tasks: normalizedTasks,
audits: Array.from(new Map(collectionItems(audits).map(normalizeAudit)
.map((entry) => [entry.eventId || `${entry.timestamp}:${entry.action}`, entry])).values()).slice(-200),
logs: [...normalizedLogs, ...taskLogs.map(normalizeLog), ...auditLogs].slice(-300),
};
}
function listTasks(query = {}) {
const all = snapshot().tasks;
const filtered = all.filter((task) => (!query.repoKey || task.repoKey === query.repoKey)
&& (!query.resourceKey || task.resourceKey === query.resourceKey)
&& (!query.state || task.state === query.state));
const offset = Math.max(0, Number.parseInt(query.offset || 0, 10) || 0);
const limit = Math.min(200, Math.max(1, Number.parseInt(query.limit || 50, 10) || 50));
return { items: filtered.slice(offset, offset + limit), total: filtered.length, offset, limit };
}
function resolveTask(taskId) {
const serviceTask = pickService(service, ['getTask', 'findTask'], taskId);
if (serviceTask) return normalizeTask(serviceTask);
return snapshot().tasks.find((task) => task.taskId === taskId) || null;
}
function controlAction(action, input = {}) {
const actor = safeText(input.actor || 'admin', 160);
const reason = safeText(input.reason || '', 1000);
if (!reason) return { ok: false, statusCode: 400, code: 'reason_required', message: '控制操作必须填写原因。' };
const rid = requestId(input.requestId);
if (state.controlRequests[rid]) {
return { ...clone(state.controlRequests[rid]), idempotent: true, overview: snapshot() };
}
const previous = snapshot();
const methodNames = {
pause: ['pause', 'pauseAll', 'setGlobalPaused'],
resume: ['resume', 'resumeAll', 'setGlobalPaused'],
disable: ['disableRepository', 'setRepositoryEnabled', 'setRepositoryStatus'],
enable: ['enableRepository', 'setRepositoryEnabled', 'setRepositoryStatus'],
cancel: ['cancelTask', 'cancelQueuedTask'],
abort: ['abortTask', 'abortTurn', 'cancelRunningTask'],
}[action] || [];
const serviceArgs = action === 'pause' || action === 'resume'
? [{ actor, reason, requestId: rid }]
: [{ ...input, actor, reason, requestId: rid }];
let result;
try {
result = pickService(service, methodNames, ...serviceArgs);
} catch (error) {
return { ok: false, statusCode: 503, code: 'workflow_service_error', message: safeText(error.message || error, 500) };
}
if (!service) return { ok: false, statusCode: 503, code: 'workflow_service_unavailable', message: 'Workflow 核心服务尚未挂接,控制操作未执行。' };
if (result === undefined) return { ok: false, statusCode: 503, code: 'workflow_control_unsupported', message: `核心服务未提供 ${action} 操作。` };
if (result && result.ok === false) {
return { ...clone(result), ok: false, statusCode: result.code === 'task_not_found' || result.code === 'repository_not_found' ? 404 : 400 };
}
const taskId = input.taskId || null;
const repoKey = input.repoKey || null;
appendAudit({
action: `admin.${action}`, actor, reason, taskId, repoKey,
fromState: taskId ? resolveTask(taskId)?.state : null,
toState: taskId ? resolveTask(taskId)?.state : null,
metadata: { requestId: rid, result: result && typeof result === 'object' ? result : null },
});
const response = { ok: true, requestId: rid, action, result: result === true ? {} : clone(result), previous: previous.summary };
state.controlRequests[rid] = response;
const requestIds = Object.keys(state.controlRequests);
while (requestIds.length > 500) delete state.controlRequests[requestIds.shift()];
persist();
return { ...clone(response), overview: snapshot() };
}
return Object.freeze({
overview: snapshot,
listTasks,
getTask(taskId) { return resolveTask(taskId); },
listAudit(query = {}) {
const items = snapshot().audits.filter((entry) => (!query.taskId || entry.taskId === query.taskId)
&& (!query.repoKey || entry.repoKey === query.repoKey));
return { items: items.slice(-Math.min(500, Number(query.limit || 200))), total: items.length };
},
control: controlAction,
configuration,
runtimeConfiguration,
updateConfiguration,
appendAudit,
setState(next) {
const cloned = clone(next) || {};
const nextState = { ...state, ...cloned };
if (cloned.settings) nextState.settings = stripGlobalConcurrency({ ...state.settings, ...cloned.settings });
state = nextState;
persist();
},
setWorkflowService(nextService) { service = nextService || null; },
getState() { return clone(state); },
});
}
function readRequestBody(req, maxBytes = 256 * 1024) {
return new Promise((resolve, reject) => {
let total = 0;
const chunks = [];
req.on('data', (chunk) => {
total += chunk.length;
if (total > maxBytes) {
reject(Object.assign(new Error('请求体过大'), { statusCode: 413 }));
req.destroy();
return;
}
chunks.push(chunk);
});
req.on('end', () => {
try {
const text = Buffer.concat(chunks).toString('utf8').trim();
resolve(text ? JSON.parse(text) : {});
} catch (error) {
reject(Object.assign(new Error('请求体不是有效 JSON'), { statusCode: 400, cause: error }));
}
});
req.on('error', reject);
});
}
function sendJson(res, statusCode, payload) {
if (res.headersSent) return;
res.writeHead(statusCode, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-cache' });
res.end(JSON.stringify(payload));
}
async function handleGiteaWorkflowManagementApi(req, res, url, management, options = {}) {
const pathname = url.pathname.replace(/\/+$/, '') || '/';
if (typeof options.authenticate === 'function' && !options.authenticate(req)) {
return sendJson(res, 401, { ok: false, code: 'unauthorized', message: 'Not authenticated' });
}
if (!management) return sendJson(res, 503, { ok: false, code: 'workflow_service_unavailable', message: 'Workflow 管理服务不可用。' });
try {
if (req.method === 'GET' && pathname === '/api/gitea-workflow/overview') return sendJson(res, 200, management.overview());
if (req.method === 'GET' && pathname === '/api/gitea-workflow/config') return sendJson(res, 200, { ok: true, ...management.configuration() });
if (req.method === 'PUT' && pathname === '/api/gitea-workflow/config') {
const body = await readRequestBody(req);
return sendJson(res, 200, { ok: true, ...management.updateConfiguration(body) });
}
if (req.method === 'GET' && pathname === '/api/gitea-workflow/tasks') {
return sendJson(res, 200, { ok: true, ...management.listTasks(Object.fromEntries(url.searchParams.entries())) });
}
const taskMatch = pathname.match(/^\/api\/gitea-workflow\/tasks\/([^/]+)$/);
if (req.method === 'GET' && taskMatch) {
const task = management.getTask(decodeURIComponent(taskMatch[1]));
return task ? sendJson(res, 200, { ok: true, task }) : sendJson(res, 404, { ok: false, code: 'task_not_found' });
}
if (req.method === 'GET' && pathname === '/api/gitea-workflow/audit') {
return sendJson(res, 200, { ok: true, ...management.listAudit(Object.fromEntries(url.searchParams.entries())) });
}
const controlMatch = pathname.match(/^\/api\/gitea-workflow\/control\/(pause|resume)$/);
if (req.method === 'POST' && controlMatch) {
const body = await readRequestBody(req);
const result = management.control(controlMatch[1], body);
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
}
const repoMatch = pathname.match(/^\/api\/gitea-workflow\/repos\/(.+)\/(disable|enable)$/);
if (req.method === 'POST' && repoMatch) {
const body = await readRequestBody(req);
const result = management.control(repoMatch[2], { ...body, repoKey: decodeURIComponent(repoMatch[1]) });
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
}
const taskControlMatch = pathname.match(/^\/api\/gitea-workflow\/tasks\/([^/]+)\/(cancel|abort)$/);
if (req.method === 'POST' && taskControlMatch) {
const body = await readRequestBody(req);
const result = management.control(taskControlMatch[2], { ...body, taskId: decodeURIComponent(taskControlMatch[1]) });
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
}
return sendJson(res, 404, { ok: false, code: 'not_found' });
} catch (error) {
return sendJson(res, error.statusCode || 500, {
ok: false,
code: error.code || 'workflow_api_error',
message: safeText(error.message || error, 500),
});
}
}
module.exports = {
createGiteaWorkflowManagement,
handleGiteaWorkflowManagementApi,
normalizeRepository,
normalizeTask,
normalizeAudit,
};

238
lib/gitea-workflow-queue.js Normal file
View File

@@ -0,0 +1,238 @@
'use strict';
/**
* Gitea Workflow 调度器:同仓库串行、跨仓库并行。
*
* 调度器只编排领域任务,不负责 clone、Codex 或 Gitea REST这些通过
* `runner(task, context)` 注入runner 返回 `{ state, ...patch }` 即可更新任务。
*/
const { EventEmitter } = require('events');
const domain = require('./gitea-workflow-domain');
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
function taskOrder(task) {
const commentCreatedAt = task?.metadata?.event?.comment?.createdAt
|| task?.metadata?.event?.comment?.created_at
|| task?.metadata?.event?.receivedAt;
const timestamp = Date.parse(commentCreatedAt || task?.createdAt || '') || Number.MAX_SAFE_INTEGER;
return `${String(timestamp).padStart(16, '0')}:${String(task?.deliveryId || task?.taskId || '')}`;
}
class GiteaWorkflowQueue extends EventEmitter {
constructor(options = {}) {
super();
if (!options.store) throw new TypeError('GiteaWorkflowQueue 需要注入 store');
this.store = options.store;
this.clock = typeof options.now === 'function' ? options.now : Date.now;
this.defaultRunner = typeof options.runner === 'function' ? options.runner : async () => ({ state: domain.TASK_STATES.SUCCEEDED });
this.pending = [];
this.active = new Map();
this.repoActive = new Map();
this.runners = new Map();
this.paused = Boolean(options.paused || this.store.getControl?.().paused);
this.draining = false;
this.started = false;
if (options.autoRecover !== false) this.recover();
}
recover() {
const tasks = this.store.recover({ maxRestartRetries: 1, now: this.clock });
this.pending = tasks.filter((task) => task.state === domain.TASK_STATES.QUEUED
|| task.state === domain.TASK_STATES.RETRY_WAIT)
.sort((a, b) => taskOrder(a).localeCompare(taskOrder(b)))
.map((task) => task.taskId);
this.started = true;
this.drain();
return tasks;
}
setRunner(taskId, runner) {
if (typeof runner === 'function') this.runners.set(String(taskId), runner);
return this;
}
enqueue(taskOrId, runner) {
const taskId = typeof taskOrId === 'string' ? taskOrId : taskOrId?.taskId;
if (!taskId) return Promise.reject(new TypeError('enqueue 缺少 taskId'));
let task = this.store.getTask(taskId);
if (!task && typeof taskOrId === 'object') task = this.store.createTask(taskOrId);
if (!task) return Promise.reject(Object.assign(new Error('找不到任务'), { code: 'task_not_found' }));
if (domain.TERMINAL_STATES.has(task.state)) return Promise.resolve({ task: this.store.getTask(taskId), skipped: true });
if (task.state !== domain.TASK_STATES.QUEUED) {
task = this.store.transitionTask(taskId, domain.TASK_STATES.QUEUED);
}
if (!this.pending.includes(taskId) && !this.active.has(taskId)) this.pending.push(taskId);
this.pending.sort((leftId, rightId) => taskOrder(this.store.getTask(leftId)).localeCompare(taskOrder(this.store.getTask(rightId))));
if (typeof runner === 'function') this.runners.set(taskId, runner);
this.emit('queued', this.store.getTask(taskId));
this.drain();
return this.waitForTerminal(taskId);
}
waitForTerminal(taskId, timeoutMs = 0) {
const existing = this.store.getTask(taskId);
if (existing && (domain.TERMINAL_STATES.has(existing.state) || existing.state === domain.TASK_STATES.WAITING_USER)) {
return Promise.resolve({ task: existing });
}
return new Promise((resolve, reject) => {
let timer = null;
const check = (task) => {
if (!task || task.taskId !== taskId) return;
if (domain.TERMINAL_STATES.has(task.state) || task.state === domain.TASK_STATES.WAITING_USER) {
cleanup(); resolve({ task });
}
};
const cleanup = () => {
this.off('updated', check);
if (timer) clearTimeout(timer);
};
this.on('updated', check);
if (timeoutMs > 0) timer = setTimeout(() => { cleanup(); reject(new Error('等待任务完成超时')); }, timeoutMs);
check(this.store.getTask(taskId));
});
}
pause(input = {}) {
this.paused = true;
this.store.setPaused(true, input);
this.emit('paused', this.store.getControl());
return this.store.getControl();
}
resume(input = {}) {
this.paused = false;
this.store.setPaused(false, input);
this.emit('resumed', this.store.getControl());
this.drain();
return this.store.getControl();
}
cancelQueued(taskId, input = {}) {
const task = this.store.getTask(taskId);
if (!task) return { ok: false, code: 'task_not_found' };
if (task.state !== domain.TASK_STATES.QUEUED && task.state !== domain.TASK_STATES.RETRY_WAIT
&& task.state !== domain.TASK_STATES.BLOCKED_WORKSPACE && task.state !== domain.TASK_STATES.WAITING_USER) {
return { ok: false, code: 'task_not_queued', task };
}
this.pending = this.pending.filter((id) => id !== String(taskId));
const updated = this.store.transitionTask(taskId, domain.TASK_STATES.CANCELLED, { errorCode: input.reason || 'cancelled_by_operator' });
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, actor: input.actor || 'admin', action: 'task.cancelled', fromState: task.state, toState: updated.state, metadata: { reason: input.reason || null } });
this.emit('updated', updated);
return { ok: true, task: updated };
}
abortRunning(taskId, input = {}) {
const entry = this.active.get(String(taskId));
const task = this.store.getTask(taskId);
if (!task) return { ok: false, code: 'task_not_found' };
if (!entry || !domain.RUNNING_STATES.has(task.state)) {
if (task.state === domain.TASK_STATES.ABORTED || task.state === domain.TASK_STATES.CANCELLED) return { ok: true, task };
return { ok: false, code: 'task_not_running', task };
}
const updated = this.store.transitionTask(taskId, domain.TASK_STATES.ABORTING, { errorCode: input.reason || 'abort_requested' });
entry.abortRequested = true;
try { entry.controller.abort(new Error(input.reason || '管理员请求中止')); } catch {}
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, actor: input.actor || 'admin', action: 'task.abort_requested', fromState: task.state, toState: updated.state, turnId: task.turnId, metadata: { reason: input.reason || null } });
this.emit('updated', updated);
return { ok: true, task: updated };
}
getStatus() {
return {
paused: this.paused,
running: this.active.size,
queued: this.pending.length,
activeTaskIds: [...this.active.keys()],
};
}
async drain() {
if (this.draining || this.paused) return;
this.draining = true;
try {
while (!this.paused) {
const index = this.pending.findIndex((id) => {
const task = this.store.getTask(id);
return task && (task.state === domain.TASK_STATES.QUEUED || task.state === domain.TASK_STATES.RETRY_WAIT)
&& !this.repoActive.has(task.repoKey || task.taskId)
&& (!task.nextRetryAt || new Date(task.nextRetryAt).getTime() <= this.clock());
});
if (index < 0) break;
const taskId = this.pending.splice(index, 1)[0];
this.run(taskId).catch(() => undefined);
}
} finally {
this.draining = false;
}
}
async run(taskId) {
const current = this.store.getTask(taskId);
if (!current) return;
const repoLockKey = current.repoKey || current.taskId;
this.repoActive.set(repoLockKey, taskId);
const controller = new AbortController();
this.active.set(taskId, { controller, abortRequested: false, startedAt: domain.iso(this.clock) });
let task = this.store.transitionTask(taskId, domain.TASK_STATES.PREPARING);
task = this.store.upsertTask({ ...task, attempt: Number(task.attempt || 0) + 1 });
this.emit('updated', task);
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, action: 'task.preparing', fromState: current.state, toState: task.state, deliveryId: task.deliveryId });
try {
task = this.store.transitionTask(taskId, domain.TASK_STATES.RUNNING);
this.emit('updated', task);
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, action: 'task.running', fromState: domain.TASK_STATES.PREPARING, toState: task.state, turnId: task.turnId });
const runner = this.runners.get(taskId) || this.defaultRunner;
const result = await runner(this.store.getTask(taskId), { signal: controller.signal, queue: this, status: this.getStatus() });
const patch = result && typeof result === 'object' ? result : {};
let desired = patch.state || (controller.signal.aborted ? domain.TASK_STATES.ABORTED : domain.TASK_STATES.SUCCEEDED);
if (patch.waitingUser) desired = domain.TASK_STATES.WAITING_USER;
if (patch.replyConfirmed) {
desired = patch.usedRestFallback
? domain.TASK_STATES.SUCCEEDED_WITH_REST_FALLBACK
: domain.TASK_STATES.SUCCEEDED;
}
if (desired === domain.TASK_STATES.ABORTED && task.state !== domain.TASK_STATES.ABORTING) {
task = this.store.transitionTask(taskId, domain.TASK_STATES.ABORTING);
}
if (desired === domain.TASK_STATES.WAITING_USER) {
task = this.store.transitionTask(taskId, desired, patch);
} else if (domain.canTransition(task.state, desired)) {
task = this.store.transitionTask(taskId, desired, patch);
} else if (task.state === domain.TASK_STATES.RUNNING
&& (desired === domain.TASK_STATES.SUCCEEDED || desired === domain.TASK_STATES.SUCCEEDED_WITH_REST_FALLBACK)) {
task = this.store.transitionTask(taskId, domain.TASK_STATES.VERIFYING_REPLY, patch);
task = this.store.transitionTask(taskId, desired, patch);
} else if (task.state === domain.TASK_STATES.RUNNING && desired === domain.TASK_STATES.FAILED_REPLY) {
task = this.store.transitionTask(taskId, domain.TASK_STATES.VERIFYING_REPLY, patch);
task = this.store.transitionTask(taskId, desired, patch);
} else {
task = this.store.transitionTask(taskId, controller.signal.aborted ? domain.TASK_STATES.ABORTED : domain.TASK_STATES.FAILED, { errorCode: 'invalid_runner_state' });
}
if (patch.turn) this.store.upsertTurn(patch.turn);
this.emit('updated', task);
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, action: `task.${task.state}`, fromState: domain.TASK_STATES.RUNNING, toState: task.state, turnId: task.turnId, errorCode: task.errorCode });
if (task.state === domain.TASK_STATES.RETRY_WAIT) {
this.pending.push(task.taskId);
}
} catch (error) {
const latest = this.store.getTask(taskId);
const target = latest.state === domain.TASK_STATES.ABORTING
? domain.TASK_STATES.ABORTED
: (error.code === 'blocked_workspace' ? domain.TASK_STATES.BLOCKED_WORKSPACE : domain.TASK_STATES.FAILED);
task = this.store.transitionTask(taskId, target, { errorCode: error.code || 'runner_failed', errorMessage: error.message || String(error) });
this.emit('updated', task);
this.store.appendAudit({ taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, action: `task.${target}`, fromState: latest.state, toState: target, errorCode: task.errorCode });
} finally {
this.active.delete(taskId);
this.repoActive.delete(repoLockKey);
this.runners.delete(taskId);
this.drain();
}
}
}
function createGiteaWorkflowQueue(options) { return new GiteaWorkflowQueue(options); }
module.exports = { GiteaWorkflowQueue, createGiteaWorkflowQueue, sleep };

View File

@@ -0,0 +1,314 @@
'use strict';
/**
* Gitea Workflow 核心编排服务。
*
* 该服务是 Webhook/管理 API 与领域存储、队列之间的适配层。server.js
* 只需在路由中注入本服务实例,即可逐步接入,不需要复制状态机逻辑。
*/
const crypto = require('crypto');
const domain = require('./gitea-workflow-domain');
const { createGiteaWorkflowStore } = require('./gitea-workflow-store');
const { createGiteaWorkflowQueue } = require('./gitea-workflow-queue');
const {
normalizeWebhookEvent,
verifyWebhookSignature,
} = require('./gitea-webhook');
function safeText(value, max = 1000) {
return typeof value === 'string' ? value.trim().slice(0, max) : '';
}
function header(headers, names) {
const source = headers || {};
for (const wanted of names) {
for (const [key, value] of Object.entries(source)) {
if (key.toLowerCase() === wanted.toLowerCase()) return Array.isArray(value) ? value[0] : value;
}
}
return '';
}
function buildGiteaMcpConfig(config, options = {}) {
const normalized = domain.normalizeConfig(config || {});
const host = safeText(options.host || normalized.gitea.host, 500);
const token = String(options.token ?? normalized.gitea.token ?? '');
const command = safeText(options.command || normalized.gitea.mcpCommand, 200) || 'gitea-mcp';
const args = Array.isArray(options.args) ? options.args.slice() : ['-t', 'stdio'];
if (host && !args.includes('-H')) args.push('-H', host);
return {
type: 'stdio',
command,
args,
env: {
...(options.env || {}),
GITEA_HOST: host,
GITEA_ACCESS_TOKEN: token,
GITEA_TOKEN: token,
},
};
}
function parseBody(rawBody) {
try {
return JSON.parse(Buffer.isBuffer(rawBody) ? rawBody.toString('utf8') : String(rawBody || ''));
} catch {
const error = new Error('Webhook 请求体不是有效 JSON');
error.code = 'invalid_json';
throw error;
}
}
class GiteaWorkflowService {
constructor(options = {}) {
this.config = domain.normalizeConfig(options.config || options, options.env || process.env);
this.store = options.store || createGiteaWorkflowStore({
filePath: options.statePath || options.filePath,
now: options.now,
});
this.clock = typeof options.now === 'function' ? options.now : Date.now;
this.runner = typeof options.runner === 'function' ? options.runner : null;
this.queue = options.queue || createGiteaWorkflowQueue({
store: this.store,
runner: this.runner || (async (task) => ({ state: domain.TASK_STATES.FAILED, errorCode: 'runner_not_configured', errorMessage: `未配置 Codex runner${task.taskId})。` })),
autoRecover: options.autoRecover !== false,
now: this.clock,
});
this.queue.on('updated', (task) => this.emitTaskAudit(task));
}
emitTaskAudit(task) {
if (!task || !task.taskId) return null;
return this.store.appendAudit({
taskId: task.taskId,
sessionKey: task.sessionKey,
repoKey: task.repoKey,
action: 'task.updated',
toState: task.state,
deliveryId: task.deliveryId,
turnId: task.turnId,
errorCode: task.errorCode,
});
}
verify(rawBody, headers = {}) {
const secret = this.config.gitea.webhookSecret;
// 内部部署可以不配置 Secret配置后仍严格执行 HMAC 验签。
if (!secret) return true;
const signature = header(headers, ['x-gitea-signature', 'x-hub-signature-256', 'x-signature']);
return Boolean(secret && verifyWebhookSignature(rawBody, signature, secret));
}
async ingestWebhook({ rawBody, headers = {} } = {}) {
if (!this.verify(rawBody, headers)) return { statusCode: 401, payload: { ok: false, code: 'invalid_signature' } };
let payload;
try { payload = parseBody(rawBody); } catch (error) { return { statusCode: 400, payload: { ok: false, code: error.code } }; }
const deliveryId = safeText(header(headers, ['x-gitea-delivery', 'x-delivery-id', 'x-github-delivery']) || payload.delivery_id, 180);
if (!deliveryId) return { statusCode: 400, payload: { ok: false, code: 'missing_delivery_id' } };
const deliveryKey = `${this.config.instanceId}:${deliveryId}`;
const claim = this.store.claimDelivery(deliveryKey, { deliveryId, metadata: { event: header(headers, ['x-gitea-event']) } });
if (claim.duplicate) {
return { statusCode: 200, payload: { ok: true, duplicate: true, deliveryId, taskId: claim.record.taskId }, record: claim.record };
}
const event = normalizeWebhookEvent({
payload,
headers,
instanceId: this.config.instanceId,
botIdentity: { login: this.config.botLogin, id: this.config.botId },
});
if (event.ignored) {
this.store.updateDelivery(deliveryKey, { status: 'ignored', reason: event.reason });
this.store.appendAudit({ action: 'webhook.ignored', actor: 'gitea', deliveryId, metadata: { reason: event.reason, eventName: event.eventName } });
return { statusCode: 200, payload: { ok: true, ignored: true, reason: event.reason, deliveryId }, event };
}
const repository = this.store.upsertRepository({
instanceId: this.config.instanceId,
owner: event.repository.owner,
name: event.repository.name,
cloneUrl: event.repository.cloneUrl,
defaultBranch: this.config.gitea.defaultBranchOverride || event.repository.defaultBranch,
});
if (repository.enabled === false || repository.status === 'disabled') {
this.store.updateDelivery(deliveryKey, { status: 'rejected', reason: 'repository_disabled' });
this.store.appendAudit({ action: 'webhook.rejected', actor: 'gitea', deliveryId, repoKey: repository.key, errorCode: 'repository_disabled' });
return { statusCode: 200, payload: { ok: true, rejected: true, reason: 'repository_disabled', deliveryId } };
}
const sessionInput = {
instanceId: this.config.instanceId,
owner: event.repository.owner,
repo: event.repository.name,
kind: event.resource.kind,
number: event.resource.number,
};
const session = this.store.upsertSession(sessionInput);
const task = this.store.createTask({
taskId: `gitea-task-${crypto.createHash('sha256').update(deliveryKey).digest('hex').slice(0, 24)}`,
deliveryKey, deliveryId, repoKey: repository.key, resourceKey: session.resourceKey,
sessionKey: session.sessionKey, commentId: event.comment.id || null,
actor: event.comment.author.login || null, instruction: event.mention.instruction,
state: domain.TASK_STATES.RECEIVED,
// 保留经过规范化的仓库/资源上下文,供工作区与 Codex runner 使用;其中不含 Token/Secret。
metadata: { eventName: event.eventName, resource: event.resource, event },
});
this.store.updateDelivery(deliveryKey, { status: 'accepted', taskId: task.taskId });
this.store.appendAudit({ action: 'webhook.accepted', actor: 'gitea', deliveryId, taskId: task.taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, fromState: domain.TASK_STATES.RECEIVED, toState: domain.TASK_STATES.QUEUED, commentId: task.commentId });
if (this.runner) this.queue.setRunner(task.taskId, this.runner);
this.queue.enqueue(task.taskId).catch(() => undefined);
return { statusCode: 202, payload: { ok: true, accepted: true, deliveryId, taskId: task.taskId, sessionKey: task.sessionKey }, task, event };
}
/**
* 接收已经由独立 Webhook receiver 验签和规范化的任务,供 server.js
* 保持最小挂接点。该方法仍在本地 store 中执行 delivery 幂等。
*/
enqueueNormalizedTask(input = {}) {
const event = input.event || input;
const deliveryId = safeText(input.deliveryId || event.deliveryId, 180);
const deliveryKey = safeText(input.deliveryKey || `${this.config.instanceId}:${deliveryId}`, 240);
const claim = this.store.claimDelivery(deliveryKey, { deliveryId, taskId: input.taskId || null, metadata: { source: 'normalized_receiver' } });
if (claim.duplicate) {
return { ok: true, duplicate: true, task: claim.record.taskId ? this.store.getTask(claim.record.taskId) : null, delivery: claim.record };
}
const repository = this.store.upsertRepository({
instanceId: this.config.instanceId,
owner: event.repository?.owner,
name: event.repository?.name,
cloneUrl: event.repository?.cloneUrl,
defaultBranch: this.config.gitea.defaultBranchOverride || event.repository?.defaultBranch,
});
if (repository.enabled === false || repository.status === 'disabled') {
this.store.updateDelivery(deliveryKey, { status: 'rejected', reason: 'repository_disabled' });
this.store.appendAudit({ action: 'webhook.rejected', actor: 'gitea', deliveryId, repoKey: repository.key, errorCode: 'repository_disabled' });
return { ok: false, duplicate: false, rejected: true, code: 'repository_disabled', repository };
}
const resource = event.resource || {};
const session = this.store.upsertSession({ instanceId: this.config.instanceId, owner: repository.owner, repo: repository.name, kind: resource.kind, number: resource.number });
const task = this.store.createTask({
taskId: input.taskId || `gitea-task-${crypto.createHash('sha256').update(deliveryKey).digest('hex').slice(0, 24)}`,
deliveryKey, deliveryId, repoKey: repository.key, resourceKey: session.resourceKey, sessionKey: session.sessionKey,
commentId: event.comment?.id || null, actor: event.comment?.author?.login || null,
instruction: event.mention?.instruction || event.comment?.body || '', metadata: { eventName: event.eventName, resource, event },
state: domain.TASK_STATES.RECEIVED,
});
this.store.updateDelivery(deliveryKey, { status: 'accepted', taskId: task.taskId });
this.store.appendAudit({ action: 'webhook.accepted', actor: 'gitea', deliveryId, taskId: task.taskId, repoKey: task.repoKey, sessionKey: task.sessionKey, commentId: task.commentId });
if (this.runner) this.queue.setRunner(task.taskId, this.runner);
this.queue.enqueue(task.taskId).catch(() => undefined);
return { ok: true, duplicate: false, task, repository, session };
}
/** 运行时挂接 Codex/工作区 runner并立即尝试领取队列。 */
setRunner(runner) {
if (typeof runner !== 'function') throw new TypeError('runner 必须是函数');
this.runner = runner;
this.queue.defaultRunner = runner;
for (const task of this.store.listTasks({ state: domain.TASK_STATES.QUEUED })) this.queue.setRunner(task.taskId, runner);
this.queue.drain();
return this;
}
/**
* 在进程内更新管理页保存的连接配置,无需重启服务。
* Secret 只从管理适配器传入内存配置,不会写入任务、日志或公开响应。
*/
updateConfig(input = {}) {
const current = this.config || {};
const currentGitea = current.gitea || {};
this.config = domain.normalizeConfig({
...current,
instanceId: input.instanceId ?? current.instanceId,
botLogin: input.botLogin ?? current.botLogin,
botId: input.botId ?? current.botId,
workspaceRoot: input.workspaceRoot ?? current.workspaceRoot,
defaultBranchOverride: input.defaultBranchOverride ?? currentGitea.defaultBranchOverride,
gitea: {
...currentGitea,
...(input.gitea || {}),
host: input.host ?? input.gitea?.host ?? currentGitea.host,
token: input.botToken ?? input.gitea?.token ?? currentGitea.token,
webhookSecret: input.webhookSecret ?? input.gitea?.webhookSecret ?? currentGitea.webhookSecret,
defaultBranchOverride: input.defaultBranchOverride ?? input.gitea?.defaultBranchOverride ?? currentGitea.defaultBranchOverride,
},
});
return this.config;
}
getOverview() {
const tasks = this.store.listTasks();
const repositories = this.store.listRepositories();
return {
ok: true,
available: true,
generatedAt: domain.iso(this.clock),
control: this.store.getControl(),
summary: {
running: tasks.filter((task) => domain.RUNNING_STATES.has(task.state)).length,
queued: tasks.filter((task) => task.state === domain.TASK_STATES.QUEUED || task.state === domain.TASK_STATES.RETRY_WAIT).length,
repositories: repositories.length,
blocked: repositories.filter((repo) => repo.dirty || repo.status === 'blocked').length,
},
repositories,
tasks,
audits: this.store.listAudits().slice(-200),
queue: this.queue.getStatus(),
};
}
listRepositories(query = {}) { return this.store.listRepositories(query); }
listTasks(query = {}) { return this.store.listTasks(query); }
getTask(taskId) { return this.store.getTask(taskId); }
listSessions(query = {}) { return this.store.listSessions(query); }
listTurns(query = {}) { return this.store.listTurns(query); }
listAudits(query = {}) { return this.store.listAudits(query); }
listAudit(query = {}) { return this.listAudits(query); }
pause(input = {}) { const result = this.queue.pause(input); this.store.appendAudit({ action: 'admin.pause', actor: input.actor || 'admin', metadata: { reason: input.reason || null } }); return result; }
resume(input = {}) { const result = this.queue.resume(input); this.store.appendAudit({ action: 'admin.resume', actor: input.actor || 'admin', metadata: { reason: input.reason || null } }); return result; }
setRepositoryEnabled(repoKey, enabled, input = {}) {
const repository = this.store.getRepository(repoKey);
if (!repository) return { ok: false, code: 'repository_not_found' };
const updated = this.store.upsertRepository({ ...repository, enabled: Boolean(enabled), status: enabled ? 'active' : 'disabled' });
this.store.appendAudit({ action: enabled ? 'admin.repository_enabled' : 'admin.repository_disabled', actor: input.actor || 'admin', repoKey, metadata: { reason: input.reason || null } });
if (enabled) {
for (const task of this.store.listTasks({ repoKey, state: domain.TASK_STATES.BLOCKED_WORKSPACE })) {
try { this.queue.enqueue(task.taskId); } catch {}
}
}
return { ok: true, repository: updated };
}
disableRepository(input = {}) { return this.setRepositoryEnabled(input.repoKey, false, input); }
enableRepository(input = {}) { return this.setRepositoryEnabled(input.repoKey, true, input); }
cancelTask(input = {}) { return this.queue.cancelQueued(input.taskId || input.id, input); }
abortTask(input = {}) { return this.queue.abortRunning(input.taskId || input.id, input); }
abortTurn(input = {}) { return this.abortTask(input); }
buildThreadConfig(options = {}) {
const session = options.session || {};
const settings = {
mode: 'yolo',
...(this.config.codex.model ? { model: this.config.codex.model } : {}),
...(this.config.codex.reasoningEffort ? { reasoning_effort: this.config.codex.reasoningEffort } : {}),
...(this.config.codex.developerInstructions ? { developer_instructions: this.config.codex.developerInstructions } : {}),
};
return {
cwd: options.cwd || null,
mcp_servers: {
gitea: buildGiteaMcpConfig(this.config, options),
},
collaborationMode: { settings },
sessionKey: session.sessionKey || null,
};
}
publicConfig() { return domain.publicConfig(this.config); }
}
function createGiteaWorkflowService(options) { return new GiteaWorkflowService(options); }
module.exports = {
GiteaWorkflowService,
buildGiteaMcpConfig,
createGiteaWorkflowService,
parseBody,
};

300
lib/gitea-workflow-store.js Normal file
View File

@@ -0,0 +1,300 @@
'use strict';
/**
* Gitea Workflow 持久化仓库。
*
* 使用单个 JSON 文件保存 MVP 领域状态,所有写入都通过同目录临时文件
* + rename 完成,进程崩溃时不会留下半截 JSON。业务层可以替换为数据库
* 但应保持本文件暴露的幂等键和查询契约。
*/
const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const domain = require('./gitea-workflow-domain');
const MAX_AUDITS = 10_000;
const MAX_DELIVERIES = 20_000;
function clone(value) {
return domain.clone(value);
}
function atomicWriteJson(filePath, value) {
const target = path.resolve(filePath);
fs.mkdirSync(path.dirname(target), { recursive: true });
const temp = `${target}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.tmp`;
fs.writeFileSync(temp, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 });
try {
fs.renameSync(temp, target);
} finally {
try { if (fs.existsSync(temp)) fs.unlinkSync(temp); } catch {}
}
}
function emptyState() {
return {
schemaVersion: 1,
updatedAt: domain.iso(),
control: { paused: false, reason: null, actor: 'system', version: 0, updatedAt: domain.iso() },
repositories: {},
sessions: {},
tasks: {},
turns: {},
deliveries: {},
audits: [],
};
}
function sanitizeMetadata(value, depth = 0) {
if (depth > 4) return '[truncated]';
if (Array.isArray(value)) return value.slice(0, 100).map((item) => sanitizeMetadata(item, depth + 1));
if (!value || typeof value !== 'object') return typeof value === 'string' ? value.slice(0, 2000) : value;
const output = {};
for (const [key, item] of Object.entries(value)) {
if (/token|secret|password|authorization|api[-_]?key/i.test(key)) {
output[key] = '[redacted]';
} else {
output[key] = sanitizeMetadata(item, depth + 1);
}
}
return output;
}
class GiteaWorkflowStore {
constructor(options = {}) {
this.filePath = options.filePath ? path.resolve(options.filePath) : null;
this.clock = typeof options.now === 'function' ? options.now : Date.now;
this.state = emptyState();
this.load();
}
load() {
if (!this.filePath || !fs.existsSync(this.filePath)) return this.state;
try {
const parsed = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
const fresh = emptyState();
this.state = {
...fresh,
...parsed,
control: { ...fresh.control, ...(parsed.control || {}) },
repositories: parsed.repositories && typeof parsed.repositories === 'object' ? parsed.repositories : {},
sessions: parsed.sessions && typeof parsed.sessions === 'object' ? parsed.sessions : {},
tasks: parsed.tasks && typeof parsed.tasks === 'object' ? parsed.tasks : {},
turns: parsed.turns && typeof parsed.turns === 'object' ? parsed.turns : {},
deliveries: parsed.deliveries && typeof parsed.deliveries === 'object' ? parsed.deliveries : {},
audits: Array.isArray(parsed.audits) ? parsed.audits : [],
};
}
} catch {
// 启动时损坏的状态文件不能让 Webhook 进程直接退出;保留空状态并在下次写入时修复。
this.state = emptyState();
}
return this.state;
}
persist() {
this.state.updatedAt = domain.iso(this.clock);
if (this.filePath) atomicWriteJson(this.filePath, this.state);
return this.snapshot();
}
snapshot() { return clone(this.state); }
getControl() { return clone(this.state.control); }
setPaused(paused, input = {}) {
const next = Boolean(paused);
const previous = this.state.control;
this.state.control = {
paused: next,
reason: typeof input.reason === 'string' ? input.reason.slice(0, 1000) : null,
actor: typeof input.actor === 'string' ? input.actor.slice(0, 160) : 'system',
version: Number(previous.version || 0) + (previous.paused === next ? 0 : 1),
updatedAt: domain.iso(this.clock),
};
this.persist();
return this.getControl();
}
upsertRepository(input) {
const key = input.key || domain.repoKeyFor(input);
const previous = this.state.repositories[key];
const record = domain.createRepositoryRecord({ ...previous, ...input, key }, { now: this.clock });
this.state.repositories[key] = record;
this.persist();
return clone(record);
}
getRepository(key) { return clone(this.state.repositories[String(key)] || null); }
listRepositories(query = {}) {
return Object.values(this.state.repositories).filter((item) => (
(!query.instanceId || item.instanceId === query.instanceId)
&& (!query.status || item.status === query.status)
&& (query.enabled === undefined ? true : item.enabled === Boolean(query.enabled))
)).map(clone);
}
upsertSession(input) {
const key = input.sessionKey || domain.sessionKeyFor(input);
const previous = this.state.sessions[key];
const record = domain.createSessionRecord({ ...previous, ...input, sessionKey: key }, { now: this.clock });
this.state.sessions[key] = record;
this.persist();
return clone(record);
}
getSession(key) { return clone(this.state.sessions[String(key)] || null); }
listSessions(query = {}) {
return Object.values(this.state.sessions).filter((item) => (
(!query.repoKey || item.resourceKey?.startsWith(String(query.repoKey)))
&& (!query.status || item.status === query.status)
)).map(clone);
}
createTask(input) {
const record = domain.createTaskRecord(input, { now: this.clock });
if (this.state.tasks[record.taskId]) return clone(this.state.tasks[record.taskId]);
this.state.tasks[record.taskId] = record;
this.persist();
return clone(record);
}
upsertTask(input) {
const id = String(input.taskId || input.id || '');
if (!id) throw new TypeError('任务缺少 taskId');
const previous = this.state.tasks[id];
const record = domain.createTaskRecord({ ...previous, ...input, taskId: id }, { now: this.clock });
this.state.tasks[id] = record;
this.persist();
return clone(record);
}
transitionTask(taskId, nextState, options = {}) {
const current = this.state.tasks[String(taskId)];
if (!current) return null;
if (options.expectedVersion !== undefined
&& Number(current.stateVersion || 0) !== Number(options.expectedVersion)) {
const error = new Error('任务版本已变化,请刷新后重试。');
error.code = 'version_conflict';
error.expectedVersion = options.expectedVersion;
error.actualVersion = current.stateVersion || 0;
throw error;
}
const next = domain.transitionTask(current, nextState, { ...options, now: options.now || this.clock });
this.state.tasks[String(taskId)] = next;
this.persist();
return clone(next);
}
getTask(taskId) { return clone(this.state.tasks[String(taskId)] || null); }
listTasks(query = {}) {
return Object.values(this.state.tasks).filter((item) => (
(!query.repoKey || item.repoKey === query.repoKey)
&& (!query.sessionKey || item.sessionKey === query.sessionKey)
&& (!query.resourceKey || item.resourceKey === query.resourceKey)
&& (!query.state && !query.status || item.state === (query.state || query.status))
)).sort((a, b) => String(a.createdAt).localeCompare(String(b.createdAt))).map(clone);
}
upsertTurn(input) {
const id = String(input.turnId || '');
if (!id) throw new TypeError('turn 缺少 turnId');
const record = { ...(this.state.turns[id] || {}), ...clone(input), turnId: id, updatedAt: domain.iso(this.clock) };
this.state.turns[id] = record;
this.persist();
return clone(record);
}
getTurn(turnId) { return clone(this.state.turns[String(turnId)] || null); }
listTurns(query = {}) {
return Object.values(this.state.turns).filter((item) => (!query.taskId || item.taskId === query.taskId)
&& (!query.sessionKey || item.sessionKey === query.sessionKey)).map(clone);
}
getDelivery(deliveryKey) { return clone(this.state.deliveries[String(deliveryKey)] || null); }
/** 原子语义由单进程事件循环保证;重复 delivery 永远返回原记录。 */
claimDelivery(deliveryKey, value = {}) {
const key = String(deliveryKey || '');
if (!key) throw new TypeError('deliveryKey 不能为空');
const existing = this.state.deliveries[key];
if (existing) return { duplicate: true, record: clone(existing) };
const record = {
deliveryKey: key,
deliveryId: value.deliveryId || key.split(':').slice(1).join(':'),
status: value.status || 'processing',
taskId: value.taskId || null,
createdAt: value.createdAt || domain.iso(this.clock),
updatedAt: value.updatedAt || domain.iso(this.clock),
metadata: sanitizeMetadata(value.metadata || {}),
};
this.state.deliveries[key] = record;
const keys = Object.keys(this.state.deliveries);
while (keys.length > MAX_DELIVERIES) delete this.state.deliveries[keys.shift()];
this.persist();
return { duplicate: false, record: clone(record) };
}
updateDelivery(deliveryKey, patch = {}) {
const key = String(deliveryKey);
if (!this.state.deliveries[key]) return null;
this.state.deliveries[key] = { ...this.state.deliveries[key], ...sanitizeMetadata(patch), updatedAt: domain.iso(this.clock) };
this.persist();
return clone(this.state.deliveries[key]);
}
appendAudit(input = {}) {
const item = {
eventId: input.eventId || crypto.randomUUID(),
taskId: input.taskId || null,
sessionKey: input.sessionKey || null,
repoKey: input.repoKey || null,
actor: input.actor || 'system',
action: input.action || 'unknown',
fromState: input.fromState || null,
toState: input.toState || null,
deliveryId: input.deliveryId || null,
turnId: input.turnId || null,
commentId: input.commentId ?? null,
errorCode: input.errorCode || null,
timestamp: input.timestamp || domain.iso(this.clock),
metadata: sanitizeMetadata(input.metadata || {}),
};
this.state.audits.push(item);
if (this.state.audits.length > MAX_AUDITS) this.state.audits.splice(0, this.state.audits.length - MAX_AUDITS);
this.persist();
return clone(item);
}
listAudits(query = {}) {
return this.state.audits.filter((item) => (!query.taskId || item.taskId === query.taskId)
&& (!query.sessionKey || item.sessionKey === query.sessionKey)
&& (!query.repoKey || item.repoKey === query.repoKey)
&& (!query.action || item.action === query.action)
&& (!query.from || String(item.timestamp) >= String(query.from))
&& (!query.to || String(item.timestamp) <= String(query.to))).map(clone);
}
recover(options = {}) {
const before = Object.values(this.state.tasks);
const after = domain.recoverTasks(before, { ...options, now: options.now || this.clock });
for (const task of after) this.state.tasks[task.taskId] = task;
this.persist();
return after.map(clone);
}
}
function createGiteaWorkflowStore(options) {
return new GiteaWorkflowStore(options);
}
module.exports = {
GiteaWorkflowStore,
atomicWriteJson,
createGiteaWorkflowStore,
emptyState,
sanitizeMetadata,
};

386
lib/gitea-workspace.js Normal file
View File

@@ -0,0 +1,386 @@
'use strict';
const crypto = require('crypto');
const fs = require('fs');
const os = require('os');
const path = require('path');
const { spawn } = require('child_process');
class WorkspaceError extends Error {
constructor(code, message, details) {
super(message);
this.name = 'WorkspaceError';
this.code = code;
if (details !== undefined) this.details = details;
}
}
class BlockedWorkspaceError extends WorkspaceError {
constructor(message, details) {
super('blocked_workspace', message || '工作区存在未提交修改,已阻止覆盖。', details);
this.name = 'BlockedWorkspaceError';
}
}
function atomicWriteJson(filePath, value) {
const target = path.resolve(filePath);
fs.mkdirSync(path.dirname(target), { recursive: true });
const tempPath = `${target}.${process.pid}.${crypto.randomBytes(5).toString('hex')}.tmp`;
fs.writeFileSync(tempPath, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 });
try { fs.renameSync(tempPath, target); } finally {
try { if (fs.existsSync(tempPath)) fs.unlinkSync(tempPath); } catch {}
}
}
function safeSegment(value, label) {
const normalized = String(value || '').trim();
if (!normalized || normalized === '.' || normalized === '..' || !/^[A-Za-z0-9._-]+$/.test(normalized)) {
throw new WorkspaceError('invalid_repository', `${label} 格式无效。`);
}
return normalized;
}
function normalizeRepository(repo) {
if (!repo || typeof repo !== 'object') throw new WorkspaceError('invalid_repository', '仓库信息无效。');
const owner = safeSegment(repo.owner, 'owner');
const name = safeSegment(repo.name, 'repo');
const cloneUrl = String(repo.cloneUrl || repo.clone_url || repo.cloneURL || repo.html_url || '').trim();
let parsed;
try { parsed = new URL(cloneUrl); } catch {
throw new WorkspaceError('invalid_clone_url', '仓库 clone URL 无效。');
}
if (parsed.protocol !== 'https:') {
throw new WorkspaceError('invalid_clone_url', '仅允许 HTTPS clone URL。');
}
parsed.username = '';
parsed.password = '';
return {
owner,
name,
fullName: `${owner}/${name}`,
cloneUrl: parsed.toString(),
defaultBranch: String(repo.defaultBranch || '').trim(),
private: repo.private === true,
};
}
// Gitea 某些版本在 Webhook 中返回 http clone_url即使实例的公开地址是
// https。仅在主机完全匹配且配置地址为 https 时升级协议,避免放宽到任意
// 不安全的远程地址。
function upgradeCloneUrlToConfiguredHttps(repo, allowedHost) {
if (!repo || !allowedHost) return repo;
const raw = String(repo.cloneUrl || repo.clone_url || repo.cloneURL || repo.html_url || '').trim();
try {
const clone = new URL(raw);
const configured = new URL(allowedHost);
if (clone.protocol === 'http:' && configured.protocol === 'https:' && clone.host === configured.host) {
clone.protocol = 'https:';
return { ...repo, cloneUrl: clone.toString() };
}
} catch {}
return repo;
}
function repoKey(instanceId, repo) {
const normalized = normalizeRepository(repo);
const instance = safeSegment(instanceId || 'default', 'instanceId');
return `${instance}:${normalized.fullName}`;
}
function createGitRunner(options = {}) {
if (typeof options.gitRunner === 'function') return options.gitRunner;
const gitCommand = String(options.gitCommand || 'git');
const maxOutputBytes = Number.isSafeInteger(options.maxOutputBytes) ? options.maxOutputBytes : 2 * 1024 * 1024;
return ({ args, cwd, env }) => new Promise((resolve, reject) => {
const child = spawn(gitCommand, args, { cwd, env, windowsHide: true });
const stdout = [];
const stderr = [];
let total = 0;
let rejected = false;
const collect = (target) => (chunk) => {
if (rejected) return;
total += chunk.length;
if (total > maxOutputBytes) {
rejected = true;
child.kill('SIGTERM');
reject(new WorkspaceError('git_output_too_large', 'Git 输出过大。'));
return;
}
target.push(chunk);
};
child.stdout.on('data', collect(stdout));
child.stderr.on('data', collect(stderr));
child.on('error', (error) => {
if (!rejected) reject(new WorkspaceError('git_spawn_failed', error.message));
});
child.on('close', (code, signal) => {
if (rejected) return;
const result = {
code: Number.isInteger(code) ? code : 1,
signal: signal || null,
stdout: Buffer.concat(stdout).toString('utf8'),
stderr: Buffer.concat(stderr).toString('utf8'),
};
if (result.code !== 0) {
const detail = String(result.stderr || result.stdout || '').trim().replace(/\s+/g, ' ').slice(-2000);
const suffix = detail ? `${detail}` : `(退出码 ${result.code}${result.signal ? `,信号 ${result.signal}` : ''}`;
const error = new WorkspaceError('git_failed', `Git 命令失败: ${args[0] || 'git'}${suffix}`, result);
reject(error);
return;
}
resolve(result);
});
});
}
async function withTempGitCredentials(options, fn) {
const token = String(options.token || '');
if (!token) throw new WorkspaceError('missing_git_token', '缺少 Gitea Bot Token。');
const username = String(options.username || 'ccweb-bot');
const tempRoot = await fs.promises.mkdtemp(path.join(os.tmpdir(), 'ccweb-git-askpass-'));
const askpassPath = path.join(tempRoot, process.platform === 'win32' ? 'askpass.cmd' : 'askpass.sh');
// 凭据只存在子进程环境和临时脚本中,绝不写入 remote URL 或普通日志。
const script = process.platform === 'win32'
? [
'@echo off',
'echo %1 | findstr /I "user" >nul',
'if %errorlevel%==0 (echo %CCWEB_GIT_USERNAME%) else (echo %CCWEB_GIT_TOKEN%)',
'',
].join('\r\n')
: [
'#!/bin/sh',
'case "$(printf %s "${1:-}" | tr "[:upper:]" "[:lower:]")" in',
' *user*) printf %s "${CCWEB_GIT_USERNAME:-}" ;;',
' *) printf %s "${CCWEB_GIT_TOKEN:-}" ;;',
'esac',
'',
].join('\n');
await fs.promises.writeFile(askpassPath, script, { mode: 0o700 });
try {
const env = {
...process.env,
...(options.env || {}),
GIT_ASKPASS: askpassPath,
GIT_TERMINAL_PROMPT: '0',
CCWEB_GIT_USERNAME: username,
CCWEB_GIT_TOKEN: token,
};
return await fn(env);
} finally {
try { await fs.promises.rm(tempRoot, { recursive: true, force: true }); } catch {}
}
}
function createRepositoryLockManager(options = {}) {
const lockRoot = path.resolve(options.lockRoot || path.join(options.workspaceRoot || os.tmpdir(), '.locks'));
const staleLockMs = Number.isSafeInteger(options.staleLockMs) && options.staleLockMs > 0 ? options.staleLockMs : 10 * 60 * 1000;
const retryMs = Number.isSafeInteger(options.retryMs) && options.retryMs > 0 ? options.retryMs : 50;
const waitTimeoutMs = Number.isSafeInteger(options.waitTimeoutMs) && options.waitTimeoutMs > 0 ? options.waitTimeoutMs : 30_000;
const localLocks = new Map();
function lockPathFor(key) {
return path.join(lockRoot, `${crypto.createHash('sha256').update(String(key)).digest('hex')}.lock`);
}
async function acquire(key) {
const normalizedKey = String(key);
const previous = localLocks.get(normalizedKey) || Promise.resolve();
let releaseLocal;
const current = new Promise((resolve) => { releaseLocal = resolve; });
const chain = previous.then(() => current);
localLocks.set(normalizedKey, chain);
await previous;
fs.mkdirSync(lockRoot, { recursive: true });
const lockPath = lockPathFor(normalizedKey);
const startedAt = Date.now();
let handle;
while (!handle) {
try {
handle = await fs.promises.open(lockPath, 'wx', 0o600);
await handle.writeFile(JSON.stringify({ pid: process.pid, key: normalizedKey, createdAt: new Date().toISOString() }));
} catch (error) {
if (error.code !== 'EEXIST') {
releaseLocal();
throw new WorkspaceError('lock_failed', `无法创建仓库锁: ${error.message}`);
}
try {
const stat = await fs.promises.stat(lockPath);
if (Date.now() - stat.mtimeMs > staleLockMs) await fs.promises.unlink(lockPath);
} catch {}
if (Date.now() - startedAt >= waitTimeoutMs) {
releaseLocal();
throw new WorkspaceError('lock_timeout', '等待仓库锁超时。', { key: normalizedKey });
}
await new Promise((resolve) => setTimeout(resolve, retryMs));
}
}
let released = false;
return async () => {
if (released) return;
released = true;
try { await handle.close(); } catch {}
try { await fs.promises.unlink(lockPath); } catch (error) {
if (error.code !== 'ENOENT') throw error;
}
releaseLocal();
if (localLocks.get(normalizedKey) === chain) localLocks.delete(normalizedKey);
};
}
async function withLock(key, fn) {
const release = await acquire(key);
try { return await fn(); } finally { await release(); }
}
return { acquire, withLock, lockPathFor };
}
function createGiteaWorkspaceManager(options = {}) {
const workspaceRoot = path.resolve(options.workspaceRoot || process.env.CC_WEB_GITEA_WORKSPACE_ROOT || process.env.WORKSPACE_ROOT || path.join(process.cwd(), 'gitea-workspaces'));
const instanceId = String(options.instanceId || process.env.CC_WEB_GITEA_INSTANCE_ID || process.env.GITEA_INSTANCE_ID || 'default');
const defaultToken = String(options.token || process.env.CC_WEB_GITEA_BOT_TOKEN || process.env.GITEA_ACCESS_TOKEN || process.env.GITEA_BOT_TOKEN || '');
const recordsPath = path.resolve(options.recordsPath || path.join(workspaceRoot, 'repositories.json'));
const botUsername = String(options.botUsername || process.env.CC_WEB_GITEA_BOT_LOGIN || process.env.GITEA_BOT_LOGIN || 'ccweb-bot');
const allowedHost = String(options.allowedHost || process.env.CC_WEB_GITEA_HOST || process.env.GITEA_HOST || '').trim().replace(/\/+$/, '');
const gitRunner = createGitRunner(options);
const locks = options.lockManager || createRepositoryLockManager({ workspaceRoot, ...options });
let records = {};
try {
const parsed = JSON.parse(fs.readFileSync(recordsPath, 'utf8'));
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) records = parsed.repositories || parsed;
} catch {}
function workspacePath(repo) {
const normalized = normalizeRepository(repo);
return path.join(workspaceRoot, safeSegment(instanceId, 'instanceId'), normalized.owner, normalized.name);
}
function persistRecords() {
atomicWriteJson(recordsPath, { schemaVersion: 1, repositories: records });
}
async function git(repoPath, args, token, extraEnv = {}) {
return withTempGitCredentials({ token, username: botUsername, env: extraEnv }, (env) => gitRunner({ args, cwd: repoPath, env }));
}
async function inspect(repoPath, token) {
const result = await git(repoPath, ['status', '--porcelain', '--untracked-files=all'], token);
const output = String(result.stdout || '');
if (output.trim()) {
const conflict = output.split('\n').some((line) => /^(UU|AA|DD|AU|UA|DU|UD)/.test(line));
throw new BlockedWorkspaceError(conflict ? '工作区存在 Git 冲突,已阻止继续。' : undefined, { status: output });
}
return { clean: true };
}
async function ensureRepository(repo, options = {}) {
const normalized = normalizeRepository(upgradeCloneUrlToConfiguredHttps(repo, allowedHost));
if (allowedHost) {
let configured;
try { configured = new URL(allowedHost); } catch { throw new WorkspaceError('invalid_gitea_host', '配置的 Gitea 地址无效。'); }
const clone = new URL(normalized.cloneUrl);
if (configured.protocol !== clone.protocol || configured.host !== clone.host) {
throw new WorkspaceError('clone_host_mismatch', '仓库 clone 地址与配置的 Gitea 实例不一致。', {
configuredHost: configured.host,
cloneHost: clone.host,
});
}
}
const key = repoKey(instanceId, normalized);
const repoPath = workspacePath(normalized);
const token = String(options.token || defaultToken || '');
return locks.withLock(key, async () => {
const existing = records[key];
const now = new Date().toISOString();
if (!existing) {
records[key] = {
key,
instanceId,
owner: normalized.owner,
name: normalized.name,
fullName: normalized.fullName,
cloneUrl: normalized.cloneUrl,
workspacePath: repoPath,
status: 'preparing',
createdAt: now,
updatedAt: now,
};
persistRecords();
}
fs.mkdirSync(path.dirname(repoPath), { recursive: true });
const gitDir = path.join(repoPath, '.git');
if (!fs.existsSync(gitDir)) {
if (fs.existsSync(repoPath) && fs.readdirSync(repoPath).length > 0) {
throw new BlockedWorkspaceError('目标工作区目录非空且不是 Git 仓库。', { workspacePath: repoPath });
}
// 工作流只需要默认分支的当前代码。浅克隆显著降低首次接入的传输量,
// 避免 Gitea/反向代理在完整历史传输期间中断;后续 fetch 仍复用同一工作区。
const cloneArgs = ['clone', '--origin', 'origin', '--depth', '1', '--single-branch'];
if (normalized.defaultBranch) cloneArgs.push('--branch', normalized.defaultBranch);
cloneArgs.push(normalized.cloneUrl, repoPath);
await git(workspaceRoot, cloneArgs, token);
} else {
if (options.allowDirty === true) {
return {
...records[key],
key,
cloneUrl: normalized.cloneUrl,
workspacePath: repoPath,
status: 'dirty_ready',
dirty: true,
dirtyReason: records[key]?.dirtyReason || '沿用当前讨论的未提交修改。',
dirtySessionKey: options.dirtySessionKey || records[key]?.dirtySessionKey || null,
repository: normalized,
};
}
await inspect(repoPath, token);
await git(repoPath, ['fetch', '--prune', 'origin'], token);
if (normalized.defaultBranch) {
// 仅在工作区确认干净后快进到配置的基线分支,绝不 reset/覆盖用户修改。
await git(repoPath, ['checkout', normalized.defaultBranch], token);
await git(repoPath, ['merge', '--ff-only', `origin/${normalized.defaultBranch}`], token);
}
}
records[key] = {
...(records[key] || {}),
key,
cloneUrl: normalized.cloneUrl,
defaultBranch: normalized.defaultBranch || records[key]?.defaultBranch || '',
workspacePath: repoPath,
status: 'ready',
dirty: false,
dirtyReason: null,
dirtySessionKey: null,
updatedAt: new Date().toISOString(),
};
persistRecords();
return { ...records[key], repository: normalized };
});
}
return {
workspaceRoot,
recordsPath,
repoKey: (repo) => repoKey(instanceId, repo),
workspacePath,
ensureRepository,
ensureWorkspace: ensureRepository,
inspectWorkspace: inspect,
checkDirty: inspect,
lockManager: locks,
withRepositoryLock: locks.withLock,
getRecord: (repo) => records[repoKey(instanceId, repo)] || null,
};
}
module.exports = {
BlockedWorkspaceError,
WorkspaceError,
createWorkspaceManager: createGiteaWorkspaceManager,
createGiteaWorkspaceManager,
createGitRunner,
createRepositoryLockManager,
normalizeRepository,
repoKey,
withTempGitCredentials,
};