AriWeb-Suite/hal/core.mjs

/**
 * hal/core.mjs — HAL 핵심 서비스: 서버·Agent 등록부, Agent 작업 배정·대기, 릴리스 등록부, 부트스트랩 상태, 지표, 포털 이벤트 outbox.
 * app.mjs(라우트)와 jobs.mjs(작업 종류)가 함께 쓴다. 모든 함수는 deps(store·vault·config·log·jobs) 를 받는다.
 */
import { hostname } from 'node:os';
import { createHash } from 'node:crypto';
import { newId, randomToken, sha256, hashPassword } from '../lib/crypto.mjs';
import { nowIso, toJson, parseJson } from '../lib/db.mjs';
import { audit } from '../lib/audit.mjs';
import { notify } from '../lib/notify.mjs';
import { redact, safeError } from '../lib/redact.mjs';
import { serviceFetch } from '../lib/serviceauth.mjs';
import { JobError } from '../lib/jobs.mjs';
import { existsSync, statSync, mkdirSync, readFileSync, writeFileSync, renameSync, unlinkSync } from 'node:fs';
import { join, basename } from 'node:path';
import { syncFromHeartbeat } from './mailsvc.mjs';

export const EXTERNAL_KINDS = new Set(['site.deploy', 'site.remove', 'site.check', 'site.rollback']);   // 외부 Agent 가 받을 수 있는 작업(§10.2)
export const SERVER_ROLES = ['web', 'mail', 'backup', 'control'];

// ------------------------------------------------------------------ 부트스트랩 상태(§32.2)
export function bootstrapState(d) { let r = d.store.get('SELECT * FROM bootstrap_state WHERE id=1'); if (!r) { d.store.run('INSERT INTO bootstrap_state(id, updated_at) VALUES(1, ?)', nowIso()); r = d.store.get('SELECT * FROM bootstrap_state WHERE id=1'); } return r; }
/** [실제 환경 결함 수정] 행(id=1)이 아직 없으면(HAL 기동 전 CLI bootstrap-release) UPDATE 가 아무 행도 바꾸지 않아 릴리스 정보가 비었다 — 먼저 행을 만든다. */
export function setBootstrap(d, patch) { bootstrapState(d); const sets = Object.keys(patch).map((k) => `${k}=?`); d.store.run(`UPDATE bootstrap_state SET ${sets.join(', ')}, updated_at=? WHERE id=1`, ...Object.values(patch), nowIso()); return bootstrapState(d); }
/** 부트스트랩 완료 여부: HAL 실행 버전 기록 + 첫 Agent 등록·승인 */
export function bootstrapComplete(d) { const b = bootstrapState(d); return !!(b.hal_running_version && b.first_agent_id && b.completed_at); }
/** HAL 기동 때: 실행 버전·인스턴스 임대 기록. 다른 인스턴스의 최근 임대가 있으면 거부(§34 "하나의 HAL 만 운영"). */
export function claimLive(d, { instanceId, version, takeover = false, now = Date.now() }) {
  const b = bootstrapState(d);
  // [통합 보완] 같은 호스트에서 이전 프로세스가 죽은 뒤(systemd 재시작·비정상 종료) 60초 임대가 남아 있으면 기동이 막힌다.
  //   임대에 호스트·PID 를 함께 기록해 두고, 같은 호스트이며 그 PID 가 더 이상 살아 있지 않으면 같은 제어 위치의 재시작으로 보아 이어받는다.
  //   다른 호스트(재해복구로 띄운 두 번째 HAL)는 여전히 거부된다(§34 "하나의 HAL 만 운영").
  const sameHostDead = b.live_host && b.live_host === hostname() && b.live_pid && !pidAlive(Number(b.live_pid));
  if (b.live_instance_id && b.live_instance_id !== instanceId && b.live_lease_at && now - Date.parse(b.live_lease_at) < 60000 && !takeover && !sameHostDead) {
    throw Object.assign(new Error(`다른 HAL 인스턴스(${b.live_instance_id})가 ${b.live_lease_at} 에 운영 중입니다. 기존 제어 위치를 먼저 중지하세요`), { code: 'another_hal_live' });
  }
  return setBootstrap(d, { live_instance_id: instanceId, live_lease_at: new Date(now).toISOString(), live_host: hostname(), live_pid: String(process.pid), hal_running_version: version, hal_started_at: nowIso() });
}
/** PID 가 살아 있는지(신호 0). 권한 오류(EPERM)는 살아 있는 것으로 본다. */
function pidAlive(pid) { if (!pid) return false; try { process.kill(pid, 0); return true; } catch (e) { return e.code === 'EPERM'; } }
export function renewLease(d, instanceId) { d.store.run('UPDATE bootstrap_state SET live_lease_at=? WHERE id=1 AND live_instance_id=?', nowIso(), instanceId); }
/** 첫 Agent 가 승인되면 부트스트랩을 닫고 수동 설치를 잠근다 */
export function maybeCompleteBootstrap(d, agentId) {
  const b = bootstrapState(d); if (b.completed_at) return b;
  if (!b.first_agent_id) setBootstrap(d, { first_agent_id: agentId, first_agent_registered_at: nowIso() });
  if (bootstrapState(d).hal_running_version) { const r = setBootstrap(d, { completed_at: nowIso(), manual_install_locked: 1, first_install: 0 }); audit(d.store, { actorType: 'system', action: 'bootstrap.complete', subjectType: 'agent', subjectId: agentId, immutable: true }); return r; }
  return bootstrapState(d);
}
/** HAL 자체 건강(§32.2 4단계) */
export function selfHealth(d) {
  const b = bootstrapState(d); let dbOk = true; try { d.store.get('SELECT 1 AS x'); } catch (_) { dbOk = false; }
  return { ok: dbOk && !!d.jobsRunning, version: d.version, uptimeSec: Math.round(process.uptime()), db: dbOk, jobs: !!d.jobsRunning, bootstrap: { complete: bootstrapComplete(d), firstInstall: !!b.first_install, releaseVersion: b.release_version, mainCommit: b.main_commit, artifactSha256: b.artifact_sha256, halRunningVersion: b.hal_running_version, firstAgent: b.first_agent_id, manualInstallLocked: !!b.manual_install_locked }, instanceId: d.instanceId };
}

// ------------------------------------------------------------------ 서버·Agent
export function createServer(d, { name, role, location = '', by = null }) {
  if (!SERVER_ROLES.includes(role)) throw new JobError('bad_role', `서버 역할은 ${SERVER_ROLES.join('/')} 중 하나입니다`);
  const id = newId('srv_'); const now = nowIso();
  d.store.run('INSERT INTO servers(id, name, role, location, status, created_at, updated_at) VALUES(?,?,?,?,\'pending\',?,?)', id, String(name).slice(0, 80), role, String(location).slice(0, 120), now, now);
  audit(d.store, { actorType: 'operator', actorId: by, action: 'server.create', subjectType: 'server', subjectId: id, detail: { name, role, location } });
  return getServer(d, id);
}
export function getServer(d, id) { const s = d.store.get('SELECT * FROM servers WHERE id=?', id); return s ? { ...s, metrics: parseJson(s.metrics_json, null), capacity: parseJson(s.capacity_json, null), facts: parseJson(s.facts_json, {}) } : null; }
export function listServers(d, { role = null } = {}) { return d.store.all(`SELECT * FROM servers ${role ? 'WHERE role=?' : ''} ORDER BY created_at`, ...(role ? [role] : [])).map((s) => publicServer(d, s)); }
/** 화면·포털용 서버 요약(비밀 없음). offline 판정은 last_seen 기준(§28.1 heartbeat). */
export function publicServer(d, s) {
  const seen = s.last_seen_at ? Date.now() - Date.parse(s.last_seen_at) : Infinity; const offline = seen > d.config.offlineAfterMs;
  const status = s.status === 'pending' ? 'pending' : offline ? 'offline' : s.status;
  const mods = d.store.all('SELECT module, target_version, actual_version, health, active, last_check_at, last_error FROM module_targets WHERE server_id=?', s.id);
  const capacity = parseJson(s.capacity_json, {}) || {}; const metrics = parseJson(s.metrics_json, null);
  return { id: s.id, name: s.name, role: s.role, location: s.location, status, rawStatus: s.status, acceptsNew: !!s.accepts_new && status === 'ready', approvedAt: s.approved_at, agentId: s.agent_id, agentVersion: s.agent_version, lastSeenAt: s.last_seen_at, stale: offline, clockSkewMs: s.clock_skew_ms,
    health: status === 'offline' ? 'down' : mods.some((m) => m.health === 'down') ? 'partial_outage' : mods.some((m) => m.health === 'degraded') ? 'degraded' : status === 'ready' ? 'ok' : 'unknown',
    metrics, capacity: { diskFreeBytes: capacity.diskFreeBytes || 0, diskTotalBytes: capacity.diskTotalBytes || 0, memFreeBytes: capacity.memFreeBytes || 0, cpuLoad: capacity.cpuLoad || 0, sitesHosted: capacity.sitesHosted || 0, reservedBytes: capacity.reservedBytes || 0, safetyMarginBytes: capacity.safetyMarginBytes || Math.round((capacity.diskTotalBytes || 0) * 0.1), exhaustEta: capacity.exhaustEta || null },
    modules: Object.fromEntries(mods.map((m) => [m.module, { target: m.target_version, actual: m.actual_version, drift: !!(m.target_version && m.actual_version && m.target_version !== m.actual_version), health: m.health, active: !!m.active, lastCheckAt: m.last_check_at, lastError: m.last_error }])),
    maintenanceNote: s.maintenance_note, isolatedReason: s.isolated_reason, facts: parseJson(s.facts_json, {}) };
}
export function updateServer(d, id, patch, by = null) {
  const allowed = ['name', 'location', 'status', 'accepts_new', 'maintenance_note', 'isolated_reason', 'role'];
  const keys = Object.keys(patch).filter((k) => allowed.includes(k)); if (!keys.length) return getServer(d, id);
  d.store.run(`UPDATE servers SET ${keys.map((k) => `${k}=?`).join(', ')}, updated_at=? WHERE id=?`, ...keys.map((k) => patch[k]), nowIso(), id);
  audit(d.store, { actorType: 'operator', actorId: by, action: 'server.update', subjectType: 'server', subjectId: id, detail: patch });
  return getServer(d, id);
}
/** 운영자 승인(§28.2 6단계): pending → ready, 신규 배치 허용 */
export function approveServer(d, id, by) { const s = getServer(d, id); if (!s) throw new JobError('server_not_found', '서버가 없습니다'); if (!s.agent_id) throw new JobError('agent_missing', 'Agent 가 아직 연결되지 않았습니다'); updateServer(d, id, { status: 'ready', accepts_new: 1 }, by); d.store.run('UPDATE servers SET approved_at=?, approved_by=? WHERE id=?', nowIso(), by, id); maybeCompleteBootstrap(d, s.agent_id); return getServer(d, id); }
/**
 * 신규 배치 가능 서버 고르기(§28.3): ready·accepts_new·여유 용량(안전 여유 제외).
 * [실제 환경 결함 수정] 서버 한 대(역할 control)로 시작하는 운영에서는 web·backup 역할 서버가 없어 사이트 배포·백업 저장을 고를 수 없었다.
 * 요청 역할의 서버가 없으면 control 서버(nginx 가 있는 관리 서버)를 web·backup 역할로도 쓴다 — 전용 서버가 생기면 그쪽이 우선한다.
 */
export function pickServer(d, { role = 'web', needBytes = 0 } = {}) {
  const fits = (s) => s.acceptsNew && s.status === 'ready' && (s.capacity.diskFreeBytes - s.capacity.reservedBytes - s.capacity.safetyMarginBytes) >= needBytes;
  let cands = listServers(d, { role }).filter(fits);
  if (!cands.length && ['web', 'backup'].includes(role)) cands = listServers(d, { role: 'control' }).filter((s) => fits(s) && (s.facts?.nginx !== false));
  cands.sort((a, b) => (b.capacity.diskFreeBytes - b.capacity.reservedBytes) - (a.capacity.diskFreeBytes - a.capacity.reservedBytes));
  return cands[0] ? { serverId: cands[0].id, name: cands[0].name } : null;
}
/** 일회용 등록 토큰(10분). 토큰 원문은 이 반환값에서 한 번만 보인다. */
export function issueRegistrationToken(d, { kind = 'managed', serverId = null, groupId = null, siteId = null, label = '', by = null, ttlMs = 10 * 60 * 1000 }) {
  if (kind === 'managed' && serverId && !getServer(d, serverId)) throw new JobError('server_not_found', '서버가 없습니다');
  const token = randomToken(32); const id = newId('rt_'); const expiresAt = new Date(Date.now() + ttlMs).toISOString();
  d.store.run('INSERT INTO registration_tokens(id, token_hash, kind, server_id, group_id, site_id, label, created_by, created_at, expires_at) VALUES(?,?,?,?,?,?,?,?,?,?)', id, sha256(token), kind, serverId, groupId, siteId, label, by, nowIso(), expiresAt);
  audit(d.store, { actorType: by ? 'operator' : 'system', actorId: by, action: 'agent.token_issue', subjectType: kind === 'external' ? 'site' : 'server', subjectId: siteId || serverId, detail: { kind, label } });
  return { id, token, expiresAt, installCommand: `curl -fsSL ${d.config.publicUrl}/agent/v1/install.sh | bash -s -- --hal ${d.config.publicUrl} --token ${token}${kind === 'external' ? ` --external --site ${siteId}` : ''}` };
}
/**
 * Agent 등록(토큰 1회 사용). 비밀은 보관소에 두고 응답에서 한 번만 돌려준다.
 * @returns {{agentId, secret, serverId, kind}}
 */
export function registerAgent(d, { token, facts = {}, version = '' }) {
  const t = d.store.get('SELECT * FROM registration_tokens WHERE token_hash=?', sha256(String(token || '')));
  if (!t) throw new JobError('bad_token', '등록 토큰이 올바르지 않습니다');
  if (t.used_at) throw new JobError('token_used', '이미 사용한 등록 토큰입니다 — 새 토큰을 발급받으세요');
  if (Date.parse(t.expires_at) < Date.now()) throw new JobError('token_expired', '등록 토큰이 만료되었습니다 — 새 토큰을 발급받으세요');
  const agentId = newId('ag_'); const secret = randomToken(32); const now = nowIso();
  return d.store.tx(() => {
    const used = d.store.run('UPDATE registration_tokens SET used_at=?, used_by_agent=? WHERE id=? AND used_at IS NULL', now, agentId, t.id).changes;
    if (!used) throw new JobError('token_used', '이미 사용한 등록 토큰입니다');
    let serverId = t.server_id;
    if (t.kind === 'managed' && !serverId) { serverId = createServer(d, { name: facts.hostname || agentId, role: facts.role || 'web', location: facts.location || '' }).id; }
    d.store.run('INSERT INTO agents(id, server_id, kind, secret_hash, status, label, group_id, site_id, version, registered_at, last_seen_at, facts_json) VALUES(?,?,?,?,\'active\',?,?,?,?,?,?,?)', agentId, serverId, t.kind, sha256(secret), t.label || facts.hostname || '', t.group_id, t.site_id, version, now, now, toJson(redact(facts)));
    d.vault.set(`service:agent:${agentId}`, secret, { kind: 'agent-secret' });
    if (serverId) d.store.run('UPDATE servers SET agent_id=?, last_seen_at=?, agent_version=?, facts_json=?, updated_at=? WHERE id=?', agentId, now, version, toJson(redact(facts)), now, serverId);
    audit(d.store, { actorType: 'agent', actorId: agentId, action: 'agent.register', subjectType: t.kind === 'external' ? 'site' : 'server', subjectId: t.site_id || serverId, detail: { kind: t.kind, version, hostname: facts.hostname } });
    return { agentId, secret, serverId, kind: t.kind, siteId: t.site_id, groupId: t.group_id };
  });
}
export function getAgent(d, id) { return d.store.get('SELECT * FROM agents WHERE id=?', id) || null; }
export function revokeAgent(d, id, by = null) { d.store.run('UPDATE agents SET status=\'revoked\', revoked_at=? WHERE id=?', nowIso(), id); d.vault.remove(`service:agent:${id}`); d.store.run('UPDATE agent_jobs SET status=\'cancelled\', ended_at=? WHERE agent_id=? AND status IN (\'queued\',\'running\')', nowIso(), id); audit(d.store, { actorType: by ? 'operator' : 'system', actorId: by, action: 'agent.revoke', subjectType: 'agent', subjectId: id, immutable: true }); }
/** 외부 Agent 상태(포털 화면용, §8.1) */
export function externalAgentStatus(d, id) {
  const a = getAgent(d, id); if (!a || a.kind !== 'external') return null;
  const seen = a.last_seen_at ? Date.now() - Date.parse(a.last_seen_at) : Infinity;
  const last = d.store.get('SELECT * FROM agent_jobs WHERE agent_id=? AND kind=\'site.deploy\' ORDER BY created_at DESC LIMIT 1', id);
  const lastCheck = d.store.get('SELECT * FROM agent_jobs WHERE agent_id=? AND kind=\'site.check\' AND status=\'done\' ORDER BY created_at DESC LIMIT 1', id);
  return { id: a.id, status: a.status === 'revoked' ? 'revoked' : !a.last_seen_at ? 'pending' : seen > d.config.offlineAfterMs ? 'offline' : 'connected', lastSeenAt: a.last_seen_at, version: a.version, siteId: a.site_id, facts: parseJson(a.facts_json, {}),
    lastDeploy: last ? { id: last.id, status: last.status, at: last.ended_at || last.started_at || last.created_at, result: parseJson(last.result_json, null), error: last.error } : null,
    lastCheck: lastCheck ? { at: lastCheck.ended_at, result: parseJson(lastCheck.result_json, null) } : null };
}
/** Webmaker 인벤토리를 2 MiB HTTP 한도 아래의 순차 배치로 받아 마지막 배치에서만 공개 facts에 원자 반영한다. */
export function webmakerInventoryBatch(d, agent, body = {}) {
  if (!agent.server_id || agent.kind !== 'managed') throw new JobError('inventory_not_allowed', '관리 서버 Agent만 인벤토리를 제출할 수 있습니다');
  const scanId = String(body.scanId || ''); const offset = Number(body.offset); const itemTotal = Number(body.itemTotal); const sites = body.sites;
  if (!scanId || scanId.length > 100 || !Number.isInteger(offset) || offset < 0 || !Number.isInteger(itemTotal) || itemTotal < 0 || itemTotal > 10000 || !Array.isArray(sites) || sites.length > 1000) throw new JobError('bad_inventory_batch', '인벤토리 배치 형식이 올바르지 않습니다');
  const safeText = (v, n) => typeof v === 'string' ? v.replace(/[\u0000-\u001f\u007f]/g, '').slice(0, n) : null;
  const safeCount = (v) => Number.isSafeInteger(Number(v)) && Number(v) >= 0 ? Number(v) : 0;
  const states = new Set(['deployed', 'unpublished', 'draft', 'unknown']); const metadataStates = new Set(['ok', 'missing', 'invalid']);
  const cleanSites = sites.map((site) => {
    if (!site || typeof site !== 'object' || Array.isArray(site) || !/^[a-f0-9]{32}$/.test(String(site.id || ''))) throw new JobError('bad_inventory_batch', '사이트 항목 형식이 올바르지 않습니다');
    const deploymentState = states.has(site.deploymentState) ? site.deploymentState : 'unknown'; const metadata = metadataStates.has(site.metadata) ? site.metadata : 'missing';
    const domains = Array.isArray(site.domains) ? [...new Set(site.domains.map((x) => safeText(x, 253)?.toLowerCase()).filter(Boolean))].slice(0, 100) : [];
    return { id: site.id, title: safeText(site.title, 200), ownerId: safeText(site.ownerId, 128), status: safeText(site.status, 80), deploymentState, domains, updatedAt: safeText(site.updatedAt, 40), metadata, bytes: safeCount(site.bytes), files: safeCount(site.files) };
  });
  const metaIn = body.meta && typeof body.meta === 'object' && !Array.isArray(body.meta) ? body.meta : {}; const safeTotal = Number(metaIn.total); const meta = { dataRoot: safeText(metaIn.dataRoot, 4096), dataDir: safeText(metaIn.dataDir, 4096), sitesRoot: safeText(metaIn.sitesRoot, 4096), total: Number.isSafeInteger(safeTotal) && safeTotal >= 0 && safeTotal <= 10000 ? safeTotal : itemTotal, truncated: !!metaIn.truncated, errors: Array.isArray(metaIn.errors) ? metaIn.errors.slice(0, 1000).map((x) => safeText(x, 500)).filter(Boolean) : [], errorsTruncated: !!metaIn.errorsTruncated, scannedAt: safeText(metaIn.scannedAt, 100) };
  if (meta.total !== itemTotal || meta.truncated) throw new JobError('inventory_incomplete', '완전한 Webmaker 인벤토리만 반영할 수 있습니다');
  return d.store.tx(() => {
    d.store.run('DELETE FROM agent_inventory_staging WHERE updated_at < ?', new Date(Date.now() - 24 * 3600 * 1000).toISOString());
    let row = d.store.get('SELECT * FROM agent_inventory_staging WHERE agent_id=? AND kind=?', agent.id, 'webmaker');
    if (offset === 0) { d.store.run('INSERT INTO agent_inventory_staging(agent_id, kind, scan_id, total, next_offset, meta_json, items_json, updated_at) VALUES(?,?,?,?,0,?,?,?) ON CONFLICT(agent_id, kind) DO UPDATE SET scan_id=excluded.scan_id,total=excluded.total,next_offset=0,meta_json=excluded.meta_json,items_json=\'[]\',updated_at=excluded.updated_at', agent.id, 'webmaker', scanId, itemTotal, toJson(meta), '[]', nowIso()); row = d.store.get('SELECT * FROM agent_inventory_staging WHERE agent_id=? AND kind=?', agent.id, 'webmaker'); }
    if (!row || row.scan_id !== scanId || row.total !== itemTotal || row.next_offset !== offset) throw new JobError('inventory_batch_order', '인벤토리 배치 순서가 맞지 않습니다 — 처음부터 다시 보내세요');
    const items = parseJson(row.items_json, []); items.push(...cleanSites); if (items.length > itemTotal || offset + cleanSites.length !== items.length) throw new JobError('bad_inventory_batch', '인벤토리 항목 수가 선언값과 다릅니다'); const itemsJson = toJson(items); if (Buffer.byteLength(itemsJson) > 16 * 1024 * 1024) throw new JobError('inventory_too_large', '인벤토리 전체 크기가 안전 한도를 넘습니다');
    const complete = items.length === itemTotal; d.store.run('UPDATE agent_inventory_staging SET next_offset=?, items_json=?, updated_at=? WHERE agent_id=? AND kind=?', items.length, itemsJson, nowIso(), agent.id, 'webmaker');
    if (!complete) return { ok: true, complete: false, nextOffset: items.length };
    const inventory = { ...parseJson(row.meta_json, {}), returned: items.length, sites: items }; const aFacts = parseJson(d.store.get('SELECT facts_json FROM agents WHERE id=?', agent.id)?.facts_json, {}) || {}; aFacts.webmakerInventory = inventory;
    const sFacts = parseJson(d.store.get('SELECT facts_json FROM servers WHERE id=?', agent.server_id)?.facts_json, {}) || {}; sFacts.webmakerInventory = inventory;
    d.store.run('UPDATE agents SET facts_json=? WHERE id=?', toJson(aFacts), agent.id); d.store.run('UPDATE servers SET facts_json=?, updated_at=? WHERE id=?', toJson(sFacts), nowIso(), agent.server_id); d.store.run('DELETE FROM agent_inventory_staging WHERE agent_id=? AND kind=?', agent.id, 'webmaker');
    return { ok: true, complete: true, nextOffset: items.length };
  });
}
/** heartbeat 처리: 지표 저장, 서버 last_seen·시계 차이·모듈 실제 버전·용량 갱신 */
export function heartbeat(d, agent, body) {
  const now = nowIso(); const m = body.metrics || {}; const skew = Number.isFinite(body.clockSkewMs) ? Math.round(body.clockSkewMs) : null;
  const mergeFacts = (currentText) => { if (!body.facts) return null; const current = parseJson(currentText, {}) || {}; const incoming = redact(body.facts); if (!Object.hasOwn(incoming, 'webmakerInventory') && current.webmakerInventory) incoming.webmakerInventory = current.webmakerInventory; return toJson(incoming); };
  const agentFacts = mergeFacts(agent.facts_json); d.store.run('UPDATE agents SET last_seen_at=?, version=?, facts_json=COALESCE(?, facts_json) WHERE id=?', now, body.version || agent.version, agentFacts, agent.id);
  if (!agent.server_id) return { ok: true };
  const capacity = { diskFreeBytes: m.disk ? m.disk.total - m.disk.used : 0, diskTotalBytes: m.disk?.total || 0, memFreeBytes: m.mem ? m.mem.total - m.mem.used : 0, cpuLoad: m.load1 || 0, sitesHosted: body.sitesHosted || 0, reservedBytes: body.reservedBytes || 0, exhaustEta: body.exhaustEta || null };
  const serverRow = d.store.get('SELECT facts_json FROM servers WHERE id=?', agent.server_id); const serverFacts = mergeFacts(serverRow?.facts_json); d.store.run('UPDATE servers SET last_seen_at=?, clock_skew_ms=?, agent_version=?, metrics_json=?, capacity_json=?, facts_json=COALESCE(?, facts_json), status=CASE WHEN status=\'offline\' THEN \'ready\' ELSE status END, updated_at=? WHERE id=?', now, skew, body.version || null, toJson(m), toJson(capacity), serverFacts, now, agent.server_id);
  d.store.run('INSERT OR REPLACE INTO metrics_samples(server_id, at, cpu, mem_used, mem_total, disk_used, disk_total, inode_used, inode_total, io_read, io_write, net_in, net_out, clock_skew_ms, load1) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
    agent.server_id, now, m.cpu ?? null, m.mem?.used ?? null, m.mem?.total ?? null, m.disk?.used ?? null, m.disk?.total ?? null, m.inode?.used ?? null, m.inode?.total ?? null, m.io?.read ?? null, m.io?.write ?? null, m.net?.in ?? null, m.net?.out ?? null, skew, m.load1 ?? null);
  for (const [name, mod] of Object.entries(body.modules || {})) {
    d.store.run('INSERT INTO module_targets(module, server_id, actual_version, health, active, last_check_at, updated_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(module, server_id) DO UPDATE SET actual_version=excluded.actual_version, health=excluded.health, active=excluded.active, last_check_at=excluded.last_check_at, updated_at=excluded.updated_at',
      name, agent.server_id, mod.version || null, mod.health || 'unknown', mod.active ? 1 : 0, now, now);
  }
  // [legacy 메일 서비스] Agent 가 보고한 arimail 모듈 상태를 그 서버의 HAL 관리형 메일 서비스 행에 비춘다(행이 없으면 무관)
  try { syncFromHeartbeat(d, agent.server_id, body.modules || {}); } catch (e) { d.log && d.log('[hal] mail_services sync 실패', e.message); }
  // 시계 차이·디스크 고갈 경고(§28.1)
  if (skew != null && Math.abs(skew) > 60000) notify(d.store, { category: 'ops', severity: 'warning', incidentKey: `server:${agent.server_id}:clock`, title: `서버 시계 차이 ${Math.round(skew / 1000)}초`, link: `/ops/servers/${agent.server_id}` });
  else { try { const { resolveIncident } = d.notifyLib; resolveIncident(d.store, `server:${agent.server_id}:clock`); } catch (_) {} }
  if (m.disk && m.disk.total && m.disk.used / m.disk.total >= 0.9) notify(d.store, { category: 'ops', severity: m.disk.used / m.disk.total >= 0.95 ? 'critical' : 'warning', incidentKey: `server:${agent.server_id}:disk`, title: `서버 디스크 ${Math.round(m.disk.used / m.disk.total * 100)}% 사용`, link: `/ops/servers/${agent.server_id}` });
  return { ok: true, serverTime: now, pendingJobs: d.store.get('SELECT COUNT(*) AS n FROM agent_jobs WHERE agent_id=? AND status=\'queued\'', agent.id).n };
}
/** 오래 heartbeat 가 없는 서버를 offline 으로 표시(주기 작업). 변경된 서버 목록 반환 */
export function markOffline(d) {
  const cut = new Date(Date.now() - d.config.offlineAfterMs).toISOString();
  const rows = d.store.all('SELECT id, name FROM servers WHERE status IN (\'ready\',\'maintenance\') AND agent_id IS NOT NULL AND (last_seen_at IS NULL OR last_seen_at<?)', cut);
  for (const r of rows) { d.store.run('UPDATE servers SET status=\'offline\', updated_at=? WHERE id=?', nowIso(), r.id); notify(d.store, { category: 'incident', severity: 'critical', mandatory: false, incidentKey: `server:${r.id}:offline`, title: `서버 ${r.name} 응답 없음(Node Agent heartbeat 중단)`, link: `/ops/servers/${r.id}` }); }
  return rows;
}
export function purgeMetrics(d) { const cut = new Date(Date.now() - d.config.metricsRetentionDays * 86400000).toISOString(); return d.store.run('DELETE FROM metrics_samples WHERE at<?', cut).changes; }
export function metrics(d, serverId, { since = null, until = null, limit = 5000 } = {}) { return d.store.all('SELECT * FROM metrics_samples WHERE server_id=? AND at>=? AND at<=? ORDER BY at DESC LIMIT ?', serverId, since || new Date(Date.now() - 90 * 86400000).toISOString(), until || nowIso(), Math.min(limit, 50000)); }

// ------------------------------------------------------------------ Agent 작업
/**
 * 재진입 가능한 배정: 'dispatch' 단계에 저장된 Agent 작업이 실패·취소로 끝났으면(재시도) 새 Agent 작업을 만든다.
 * 호출자는 ctx.step('dispatch', …) 대신 이 함수를 쓴다.
 */
export async function ensureAgentJob(d, ctx, job, { stepId = 'dispatch', agentId, kind, payload, title = '대상 Agent 에 작업 배정' }) {
  const prev = ctx.steps().find((x) => x.id === stepId);
  if (prev && prev.state === 'done' && prev.result?.agentJobId) { const aj = getAgentJob(d, prev.result.agentJobId); if (aj && ['failed', 'cancelled'].includes(aj.status)) { prev.state = 'pending'; delete prev.result; } }
  return ctx.step(stepId, () => ({ agentId, agentJobId: dispatchAgentJob(d, { halJobId: job.id, agentId, kind, payload, dedupeKey: `${job.id}:${kind}:${agentId}:${job.attempt}` }) }), title);
}
/** Agent 작업 만들기(HAL 작업의 자식). 같은 hal_job_id+kind+agent 의 미완료 작업이 있으면 재사용(재진입 멱등). */
export function dispatchAgentJob(d, { halJobId, agentId, kind, payload, dedupeKey = null }) {
  const agent = getAgent(d, agentId); if (!agent || agent.status !== 'active') throw new JobError('agent_unavailable', 'Agent 가 없거나 해지되었습니다');
  if (agent.kind === 'external' && !EXTERNAL_KINDS.has(kind)) throw new JobError('kind_not_allowed_external', `외부 Agent 는 ${kind} 작업을 받을 수 없습니다`);
  if (agent.kind === 'external' && payload?.siteId && agent.site_id && payload.siteId !== agent.site_id) throw new JobError('site_mismatch', '이 외부 Agent 는 다른 사이트에 귀속되어 있습니다');
  const key = dedupeKey || `${halJobId}:${kind}:${agentId}`;
  const ex = d.store.get('SELECT * FROM agent_jobs WHERE hal_job_id=? AND kind=? AND agent_id=? AND json_extract(payload_json, \'$._dedupe\')=?', halJobId, kind, agentId, key);
  if (ex) return ex.id;
  const id = newId('aj_');
  d.store.run('INSERT INTO agent_jobs(id, hal_job_id, agent_id, kind, payload_json, status, created_at, steps_json, log_json) VALUES(?,?,?,?,?,\'queued\',?,\'[]\',\'[]\')', id, halJobId, agentId, kind, toJson({ ...payload, _dedupe: key }), nowIso());
  return id;
}
export function getAgentJob(d, id) { const r = d.store.get('SELECT * FROM agent_jobs WHERE id=?', id); return r ? { ...r, payload: parseJson(r.payload_json, {}), steps: parseJson(r.steps_json, []), result: parseJson(r.result_json, null), log: parseJson(r.log_json, []) } : null; }
/** Agent 가 다음 작업을 가져간다(long-poll 은 라우트가 반복 호출). queued 중 가장 오래된 것 하나 → running */
export function nextAgentJob(d, agentId) {
  const r = d.store.get('SELECT id FROM agent_jobs WHERE agent_id=? AND status=\'queued\' ORDER BY created_at LIMIT 1', agentId); if (!r) return null;
  const got = d.store.run('UPDATE agent_jobs SET status=\'running\', assigned_at=?, started_at=?, heartbeat_at=? WHERE id=? AND status=\'queued\'', nowIso(), nowIso(), nowIso(), r.id).changes;
  if (!got) return null; const j = getAgentJob(d, r.id); delete j.payload._dedupe; return { id: j.id, kind: j.kind, payload: j.payload, halJobId: j.hal_job_id };
}
/** Agent 재시작 뒤 이어서 할 작업 목록(running 으로 남은 것) */
export function unfinishedAgentJobs(d, agentId) { return d.store.all('SELECT id, kind, payload_json, steps_json FROM agent_jobs WHERE agent_id=? AND status=\'running\'', agentId).map((r) => ({ id: r.id, kind: r.kind, payload: (() => { const p = parseJson(r.payload_json, {}); delete p._dedupe; return p; })(), steps: parseJson(r.steps_json, []) })); }
export function agentJobProgress(d, id, agentId, { progress, steps, log }) { const r = d.store.run('UPDATE agent_jobs SET progress=COALESCE(?, progress), steps_json=COALESCE(?, steps_json), log_json=COALESCE(?, log_json), heartbeat_at=? WHERE id=? AND agent_id=? AND status=\'running\'', progress ?? null, steps ? toJson(redact(steps)) : null, log ? toJson(redact(log)) : null, nowIso(), id, agentId).changes; return !!r; }
export function agentJobResult(d, id, agentId, { status, result, error, errorCode, userMessage, steps, log }) {
  if (!['done', 'failed', 'needs_user'].includes(status)) throw new JobError('bad_status', 'Agent 결과 상태는 done/failed/needs_user 중 하나입니다');
  const ch = d.store.run('UPDATE agent_jobs SET status=?, result_json=?, error=?, error_code=?, user_message=?, steps_json=COALESCE(?, steps_json), log_json=COALESCE(?, log_json), progress=CASE WHEN ?=\'done\' THEN 100 ELSE progress END, ended_at=? WHERE id=? AND agent_id=? AND status=\'running\'',
    status, toJson(redact(result ?? null)), error ? safeError({ message: error }) : null, errorCode || null, userMessage || null, steps ? toJson(redact(steps)) : null, log ? toJson(redact(log)) : null, status, nowIso(), id, agentId).changes;
  return !!ch;
}
export const isAgentJobCancelled = (d, id) => !!d.store.get('SELECT cancel_requested FROM agent_jobs WHERE id=?', id)?.cancel_requested;
/**
 * HAL 작업(lib/jobs run) 안에서 Agent 작업이 끝날 때까지 기다린다. 진행률·단계를 HAL 작업에 비춘다.
 * Agent 가 오래 응답하지 않으면(heartbeat 5분) 계속 기다리되 사유를 기록하고, 서버가 offline 이면 needs_user 로 멈춘다(운영자가 서버를 살리고 '이어서 진행').
 */
export async function awaitAgentJob(d, ctx, agentJobId, { timeoutMs = d.config.agentWaitTimeoutMs || 6 * 3600 * 1000, pollMs = d.config.agentPollMs || 1000 } = {}) {
  const t0 = Date.now(); let lastProgress = -1;
  while (true) {
    const j = getAgentJob(d, agentJobId); if (!j) throw new JobError('agent_job_missing', 'Agent 작업 기록이 사라졌습니다');
    if (j.status === 'done') return j.result;
    if (j.status === 'failed') throw new JobError(j.error_code || 'agent_failed', j.error || 'Agent 작업 실패', { retryable: j.error_code === 'transient' });
    if (j.status === 'needs_user') ctx.needsUser(j.user_message || 'Agent 가 사용자/운영자 확인을 요청합니다', [{ id: 'resume', label: '확인 후 이어서 진행' }]);
    if (j.status === 'cancelled') throw new JobError('agent_job_cancelled', 'Agent 작업이 취소되었습니다');
    try { ctx.checkCancel(); } catch (e) { d.store.run('UPDATE agent_jobs SET cancel_requested=1 WHERE id=?', agentJobId); throw e; }
    if (j.progress !== lastProgress) { lastProgress = j.progress; ctx.progress(j.progress); }
    const agent = getAgent(d, j.agent_id); const srv = agent?.server_id ? getServer(d, agent.server_id) : null;
    const silent = agent?.last_seen_at ? Date.now() - Date.parse(agent.last_seen_at) : Infinity;
    if (silent > 10 * 60 * 1000) ctx.needsUser(`대상 서버(${srv?.name || agent?.label || j.agent_id})의 Agent 가 10분 넘게 응답하지 않습니다. 서버 상태를 확인한 뒤 이어서 진행하세요`, [{ id: 'resume', label: 'Agent 복구 후 이어서 진행' }]);
    if (Date.now() - t0 > timeoutMs) throw new JobError('agent_timeout', 'Agent 작업이 제한 시간을 넘겼습니다');
    await new Promise((r) => setTimeout(r, pollMs));
  }
}

// ------------------------------------------------------------------ 릴리스(§32.1)
/** 키 순서에 무관한 비교용 JSON(릴리스 재등록의 "같은 내용" 판정) */
function canonJson(v) { if (Array.isArray(v)) return '[' + v.map(canonJson).join(',') + ']'; if (v && typeof v === 'object') return '{' + Object.keys(v).sort().map((k) => JSON.stringify(k) + ':' + canonJson(v[k])).join(',') + '}'; return JSON.stringify(v === undefined ? null : v); }
/**
 * 릴리스 등록(upsert). 0.2.0 검토 수정 3차 — **승인된 버전은 불변**: 같은 내용(main 커밋·빌드 시각·artifact·checks, 메모는 준 경우에만)의 재등록은 멱등하게 그대로 돌려주고(approve 요청이어도 이미 승인),
 * 어느 하나라도 다른 재등록은 approve 여부와 무관하게 `release_immutable` 로 거절하며 아무것도 바꾸지 않는다(새 내용은 새 버전 번호로). 승인되지 않은 버전의 내용 변경 재등록은 내용을 갱신하되
 * 상태를 registered 로 두고(승인자·시각 비움) 승인을 물려받지 않는다 — approve 가 참이면 releaseApprovalProblem 규칙을 지나야만 승인된다. CLI(hal/index.mjs register-release·scripts/release.mjs register)도 이 함수를 쓴다.
 */
/** 승인된 릴리스를 다른 내용으로 바꾸려는 등록인지(같은 내용이면 null). registerRelease·installArtifacts 가 같은 규칙을 쓴다 */
export function releaseImmutableProblem(ex, { mainCommit, builtAt, artifacts, checks = null, notes = null }) {
  if (!ex || ex.status !== 'approved') return null;
  const same = ex.main_commit === mainCommit && ex.built_at === builtAt && canonJson(ex.artifacts) === canonJson(artifacts || {}) && canonJson(ex.checks) === canonJson(checks) && (notes == null || notes === ex.notes);
  return same ? null : `승인된 릴리스 ${ex.version} 은 바꿀 수 없습니다 — artifact·검사 기록·main 커밋·빌드 시각·메모가 다른 빌드는 새 버전 번호로 등록하세요`;
}
/**
 * artifact 설치(0.2.0 검토 수정 4차 — 불변·fail-closed). 등록(registerRelease) **전** 에 부른다.
 *  1) 원본 파일마다 바이트의 sha256 과 크기를 manifest 의 선언값과 대조한다 — 다르면 `artifact_hash_mismatch` 로 거절(아무것도 쓰지 않는다).
 *  2) 같은 버전이 이미 승인되어 있으면 메타데이터(releaseImmutableProblem)와 설치된 파일의 해시까지 같을 때만 멱등(복사하지 않음); 하나라도 다르면 `release_immutable` 로 거절하고 기존 파일·DB 를 건드리지 않는다.
 *  3) 새 버전·미승인 버전은 대상 폴더의 임시 파일(.tmp-…)에 쓰고 해시를 다시 확인한 뒤 rename 으로 승격한다(원자적) — 승인된 파일은 어떤 경로로도 잘리거나 덮이지 않는다.
 * @param {object} d  HAL deps
 * @param {{version:string, mainCommit:string, builtAt:string, checks?:object, notes?:string, artifacts:Record<string,{file:string, sha256:string, size?:number, src:string}>}} m  src = 원본 파일 경로
 * @returns {Record<string,{file:string, sha256:string, size:number}>} 등록에 넘길 artifact 목록(파일 이름만)
 */
export function installArtifacts(d, m) {
  const entries = Object.entries(m.artifacts || {});
  const declared = {}; const verified = {};
  for (const [mod, a] of entries) {
    if (!a || !a.file || !a.src) throw new JobError('artifact_missing', `${mod}: artifact 파일 정보가 없습니다`);
    if (!existsSync(a.src)) throw new JobError('artifact_missing', `${mod}: artifact 파일이 없습니다 (${basename(a.src)})`);
    const bytes = readFileSync(a.src); const sha = createHash('sha256').update(bytes).digest('hex');
    if (String(a.sha256 || '').toLowerCase() !== sha) throw new JobError('artifact_hash_mismatch', `${mod}: artifact 해시 불일치 — manifest ${String(a.sha256 || '').slice(0, 12)}… / 파일 ${sha.slice(0, 12)}… (설치·등록하지 않습니다)`);
    if (a.size !== undefined && a.size !== null && Number(a.size) !== bytes.length) throw new JobError('artifact_hash_mismatch', `${mod}: artifact 크기 불일치 — manifest ${a.size} / 파일 ${bytes.length}`);
    declared[mod] = { file: basename(a.file), sha256: sha, size: bytes.length }; verified[mod] = { bytes, ...declared[mod], src: a.src };
  }
  const ex = getRelease(d, m.version);
  const dir = join(artifactDir(d), m.version);
  if (ex && ex.status === 'approved') {
    const why = releaseImmutableProblem(ex, { mainCommit: m.mainCommit, builtAt: m.builtAt, artifacts: declared, checks: m.checks ?? null, notes: m.notes ?? null });
    if (why) throw new JobError('release_immutable', why);
    for (const [mod, v] of Object.entries(verified)) {   // 설치된 바이트도 같아야 멱등 — 아니면 승인된 파일을 건드리지 않고 거절
      const p = join(dir, v.file); if (!existsSync(p)) throw new JobError('release_immutable', `승인된 릴리스 ${m.version} 의 ${mod} artifact 파일이 없습니다 — 설치 상태를 운영자가 확인하세요`);
      const cur = createHash('sha256').update(readFileSync(p)).digest('hex'); if (cur !== v.sha256) throw new JobError('release_immutable', `승인된 릴리스 ${m.version} 의 ${mod} artifact 는 바꿀 수 없습니다(설치된 파일과 바이트가 다름)`);
    }
    return declared;   // 멱등 — 아무것도 쓰지 않는다
  }
  mkdirSync(dir, { recursive: true });
  for (const [mod, v] of Object.entries(verified)) {
    const target = join(dir, v.file);
    if (existsSync(target) && createHash('sha256').update(readFileSync(target)).digest('hex') === v.sha256) continue;   // 이미 같은 바이트
    const tmp = join(dir, `.${v.file}.tmp-${process.pid}-${Math.random().toString(16).slice(2)}`);
    try { writeFileSync(tmp, v.bytes, { mode: 0o644 }); const again = createHash('sha256').update(readFileSync(tmp)).digest('hex'); if (again !== v.sha256) throw new JobError('artifact_hash_mismatch', `${mod}: 임시 파일 해시 불일치`); renameSync(tmp, target); }
    catch (e) { try { unlinkSync(tmp); } catch (_) {} throw e; }
  }
  return declared;
}
export function registerRelease(d, { version, mainCommit, builtAt, artifacts, checks = null, notes = null, approve = false, by = null }) {
  const now = nowIso(); const ex = getRelease(d, version);
  if (ex) {
    const same = ex.main_commit === mainCommit && ex.built_at === builtAt && canonJson(ex.artifacts) === canonJson(artifacts || {}) && canonJson(ex.checks) === canonJson(checks) && (notes == null || notes === ex.notes);
    if (ex.status === 'approved') {
      const why = releaseImmutableProblem(ex, { mainCommit, builtAt, artifacts, checks, notes }); if (why) throw new JobError('release_immutable', why);
      audit(d.store, { actorType: by ? 'operator' : 'system', actorId: by, action: 'release.register', subjectType: 'release', subjectId: version, detail: { mainCommit, artifacts: Object.keys(artifacts || {}), idempotent: true, approved: true } });
      return ex;
    }
    if (!same) d.store.run('UPDATE releases SET main_commit=?, built_at=?, artifacts_json=?, checks_json=?, notes=COALESCE(?, notes), status=\'registered\', approved_by=NULL, approved_at=NULL WHERE version=?', mainCommit, builtAt, toJson(artifacts), toJson(checks), notes, version);
  } else d.store.run('INSERT INTO releases(version, main_commit, built_at, artifacts_json, checks_json, status, notes, registered_at) VALUES(?,?,?,?,?,\'registered\',?,?)', version, mainCommit, builtAt, toJson(artifacts), toJson(checks), notes, now);
  if (approve) approveRelease(d, version, by);
  audit(d.store, { actorType: by ? 'operator' : 'system', actorId: by, action: 'release.register', subjectType: 'release', subjectId: version, detail: { mainCommit, artifacts: Object.keys(artifacts || {}) } });
  return getRelease(d, version);
}
/**
 * 릴리스 검사 기록이 승인 가능한지(0.2.0 검토 수정 2차 — fail-closed). 세 값이 **정확히** 있어야 한다: testsPassed === true(--skip-tests 빌드의 null·없음은 불가), dirty === false(작업 트리 깨끗함이 기록됨),
 * provenanceClean === true(도입 소스 출처가 깨끗하다고 확인됨). 기록이 없거나(옛 빌드) 명시적 false 가 아닌 "모름" 도 승인하지 않는다. 승인 경로(approveRelease·registerRelease approve:true·운영자 API·release.mjs register --approve)가 모두 이 한 규칙을 쓴다.
 * @returns {string|null} 승인할 수 없는 이유(없으면 null)
 */
export function releaseApprovalProblem(checks) {
  const c = checks && typeof checks === 'object' ? checks : null;
  if (!c) return '검사 기록(checks)이 없는 릴리스는 승인할 수 없습니다 — 테스트·작업 트리·도입 소스 출처가 확인된 빌드로 다시 등록하세요';
  if (c.testsPassed !== true) return `테스트 통과가 기록되지 않은 릴리스는 승인할 수 없습니다 (checks.testsPassed=${JSON.stringify(c.testsPassed ?? null)} — --skip-tests 빌드는 등록만 되고 승인은 불가)`;
  if (c.dirty !== false) return `작업 트리가 깨끗하다고 기록되지 않은 릴리스는 승인할 수 없습니다 (checks.dirty=${JSON.stringify(c.dirty ?? null)} — main 에 병합된 커밋에서 빌드하세요)`;
  if (c.provenanceClean !== true) return `도입 소스 출처(provenance)가 깨끗하다고 확인되지 않은 릴리스는 승인할 수 없습니다 (checks.provenanceClean=${JSON.stringify(c.provenanceClean ?? null)}${(c.provenanceProblems || []).length ? ': ' + c.provenanceProblems.join(' / ') : ''})`;
  return null;
}
export function approveRelease(d, version, by) { const r = getRelease(d, version); if (!r) throw new JobError('release_not_found', '릴리스가 없습니다'); const why = releaseApprovalProblem(r.checks); if (why) throw new JobError('release_checks_failed', why); d.store.run('UPDATE releases SET status=\'approved\', approved_by=?, approved_at=? WHERE version=?', by, nowIso(), version); audit(d.store, { actorType: 'operator', actorId: by, action: 'release.approve', subjectType: 'release', subjectId: version, immutable: true }); return getRelease(d, version); }
export function getRelease(d, version) { const r = d.store.get('SELECT * FROM releases WHERE version=?', version); return r ? { ...r, artifacts: parseJson(r.artifacts_json, {}), checks: parseJson(r.checks_json, null) } : null; }
export function listReleases(d) { return d.store.all('SELECT * FROM releases ORDER BY registered_at DESC').map((r) => ({ ...r, artifacts: parseJson(r.artifacts_json, {}), checks: parseJson(r.checks_json, null) })); }
export const artifactDir = (d) => join(d.config.dataDir, 'artifacts');
/** artifact 파일 경로·존재 확인 */
export function artifactPath(d, version, module) { const r = getRelease(d, version); const a = r && r.artifacts[module]; if (!a) return null; const p = join(artifactDir(d), version, a.file); return existsSync(p) ? { path: p, sha256: a.sha256, size: statSync(p).size } : null; }

// ------------------------------------------------------------------ 전송(transfer)
export function transferDir(d) { const p = join(d.config.dataDir, 'transfers'); mkdirSync(p, { recursive: true, mode: 0o700 }); return p; }
export function createTransfer(d, { id = newId('tr_'), agentId, sha256: sum, size, purpose = '', meta = null }) { const path = join(transferDir(d), id + '.bin'); d.store.run('INSERT OR REPLACE INTO transfers(id, created_by_agent, path, size_bytes, sha256, created_at, expires_at, purpose, meta_json) VALUES(?,?,?,?,?,?,?,?,?)', id, agentId, path, size, sum, nowIso(), new Date(Date.now() + d.config.transferTtlMs).toISOString(), purpose, toJson(meta)); return { id, path }; }
export function getTransfer(d, id) { const t = d.store.get('SELECT * FROM transfers WHERE id=? AND deleted_at IS NULL', id); return t && existsSync(t.path) ? t : null; }

// ------------------------------------------------------------------ 포털 이벤트 outbox(§6)
export function queuePortalEvent(d, payload) { d.store.run('INSERT INTO portal_events_outbox(id, payload_json, created_at, next_at) VALUES(?,?,?,?)', newId('ev_'), toJson(redact(payload)), nowIso(), nowIso()); }
/** outbox 전송(주기). 실패는 지수 백오프로 재시도하고 치명적으로 다루지 않는다. */
export async function flushPortalEvents(d, { fetchImpl = fetch } = {}) {
  const secret = d.vault.get('service:hal'); if (!secret) return 0;
  const rows = d.store.all('SELECT * FROM portal_events_outbox WHERE delivered_at IS NULL AND next_at<=? ORDER BY created_at LIMIT 50', nowIso()); let n = 0;
  for (const r of rows) {
    try { const res = await serviceFetch(d.config.portalUrl, 'hal', secret, 'POST', '/internal/hal-events', parseJson(r.payload_json, {}), { timeoutMs: 8000, fetchImpl }); if (!res.ok) throw new Error(`portal ${res.status} ${res.json?.error || ''}`); d.store.run('UPDATE portal_events_outbox SET delivered_at=? WHERE id=?', nowIso(), r.id); n++; }
    catch (e) { const attempts = r.attempts + 1; d.store.run('UPDATE portal_events_outbox SET attempts=?, next_at=?, last_error=? WHERE id=?', attempts, new Date(Date.now() + Math.min(600000, 5000 * 2 ** attempts)).toISOString(), safeError(e), r.id); }
  }
  d.store.run('DELETE FROM portal_events_outbox WHERE delivered_at IS NOT NULL AND delivered_at<?', new Date(Date.now() - 7 * 86400000).toISOString());
  return n;
}
export function ensureOpsAccountExists(d) { return d.store.get('SELECT COUNT(*) AS n FROM ops_accounts').n > 0; }
export { hashPassword };