dispatcher.js

'use strict';
// Canonical event-kernel dispatcher for lifecycle, tool, compact, and policy events.
// Host adapters call this module; route facts live in dispatcher-routes.json.

const path = require('path');
const fs = require('fs');
const crypto = require('crypto');
const { PROJECT_ROOT } = require('../proxy/shared');
const { appendHookExec } = require('../hooks/hook_report');
const { shouldSkipForNestedHooks } = require('../hooks/cwd_guard');
const { preWriteCheck, toHookResponse } = require('../proxy/pre_write_check');
const { isWriteFamilyTool } = require('../proxy/edit_validation');
const stateClient = require('../proxy/session_state_client');
const { normalize } = require('./envelope');
const { recordFailure } = require('../proxy/turn_failure_state');
const { spawnFileInput } = require('./fs_ipc');
const { recordHookCheckpoint } = require('./hook_decision_log');
const nativeHooks = require('./native_hooks');
const { applyOmoLive, observeOmoShadow } = require('../omo_bridge/shadow_runtime');
const { UNIVERSAL_HOOK_ABI } = require('../omo_bridge/universal_event');
const { isStrictMode } = require('../proxy/strict_mode');
const routeRegistry = require('./route_registry');
const stateRegistry = require('../proxy/state_registry');
const { renderDeny, renderInstruct, renderPolicyAggregate, renderPolicyFailure } = require('./decision_renderer');

const RETRY_STATE = path.join(PROJECT_ROOT, 'tools', 'HME', 'runtime', 'tool-retry-guard.json');
const RETRY_LOG = path.join(PROJECT_ROOT, 'tools', 'HME', 'runtime', 'tool-retry-guard.jsonl');
const INSPECTION_TOOLS = new Set(['Read', 'Grep', 'Glob']);

// Declared routing contract: non-derivable route facts (policyContext per event).
const policyContext = routeRegistry.policyContext;

function _stable(value) {
  if (Array.isArray(value)) return value.map(_stable);
  if (!value || typeof value !== 'object') return value;
  return Object.fromEntries(Object.keys(value).sort().map((key) => [key, _stable(value[key])]));
}

function _attemptDigest(tool, input) {
  return crypto.createHash('sha256').update(JSON.stringify({ tool, input: _stable(input || {}) })).digest('hex').slice(0, 16);
}

function _readRetryState() {
  return stateRegistry.read('statefile_tool_retry_guard') || {};
}

function _writeRetryState(state) {
  stateRegistry.write('statefile_tool_retry_guard', state || {});
}

function _logRetry(row) {
  stateRegistry.append('statefile_tool_retry_guard_log', { ts: new Date().toISOString(), ...row });
}

function _retryBlock(tool, input) {
  const state = _readRetryState();
  if (INSPECTION_TOOLS.has(tool)) {
    if (state.last_failed_attempt) {
      state.last_failed_attempt.recovery_tool = tool;
      state.last_failed_attempt.recovered_at = new Date().toISOString();
      _writeRetryState(state);
    }
    return null;
  }
  const digest = _attemptDigest(tool, input);
  const last = state.last_failed_attempt || {};
  if (last.digest !== digest || last.recovered_at) return null;
  const reason = `BLOCKED: repeated failed ${tool} attempt without an intervening Read/Grep/Glob or changed input.`;
  _logRetry({ decision: 'block', tool, digest, reason });
  return { stdout: renderDeny('PreToolUse', reason), stderr: ' ', exit_code: 0 };
}

function _recordToolFailure(tool, input, response) {
  const state = _readRetryState();
  const digest = _attemptDigest(tool, input);
  state.last_failed_attempt = {
    tool,
    digest,
    input,
    reason: response && (response.stderr || response.error || response.message || JSON.stringify(response).slice(0, 300)),
    ts: new Date().toISOString(),
  };
  _writeRetryState(state);
  _logRetry({ decision: 'record_failure', tool, digest, reason: state.last_failed_attempt.reason });
}

async function _recordLifecycleState(eventName, stdinJson) {
  const env = normalize(stdinJson);
  const sid = env.session_id || '';
  if (eventName === 'SessionStart') await stateClient.call('read', sid);
  if (eventName === 'UserPromptSubmit') await stateClient.call('phase', sid, { phase: 'observe', meta: { event: eventName } });
  if (eventName === 'Stop') await stateClient.call('phase', sid, { phase: 'verify', meta: { event: eventName } });
}

async function _recordPostToolEvidence(stdinJson) {
  const env = normalize(stdinJson);
  const tool = env.tool_name || '';
  const input = env.tool_input || {};
  const response = env.tool_response || {};
  if (tool !== 'Bash' && tool !== 'Read') return;
  const command = tool === 'Bash' ? String(input.command || '') : `Read ${input.file_path || ''}`;
  const excerpt = typeof response === 'string'
    ? response.slice(0, 500)
    : JSON.stringify(response).slice(0, 500);
  const exitCode = Number.isInteger(response.exit_code) ? response.exit_code : null;
  const failed = response && (response.is_error === true || response.error === true || (exitCode !== null && exitCode !== 0));
  if (failed) recordFailure(PROJECT_ROOT, { tool, reason: response.stderr || response.error || `exit ${exitCode}`, command, session_id: env.session_id || '' });
  await stateClient.call('verification-evidence', env.session_id || '', {
    session_id: env.session_id || '',
    command,
    exit_code: exitCode,
    excerpt,
    artifact: input.file_path || '',
    source: `PostToolUse:${tool}`,
  });
}

const HOOKS_DIR = path.join(PROJECT_ROOT, 'tools', 'HME', 'hooks');

function _hookScript(rel) {
  return path.join(HOOKS_DIR, rel);
}

function _scriptMapFor(eventName) {
  const raw = routeRegistry.shellByTool(eventName);
  const out = {};
  for (const [tool, scripts] of Object.entries(raw)) out[tool] = scripts.map(_hookScript);
  return out;
}

// Tool-name -> hook scripts, sourced from dispatcher-routes.json.
const PRETOOL_SCRIPTS = _scriptMapFor('PreToolUse');
const UNIVERSAL_POSTTOOL = routeRegistry.universalScripts('PostToolUse').map(_hookScript);
const POSTTOOL_SCRIPTS = _scriptMapFor('PostToolUse');
const HME_PRIMER_SCRIPT = routeRegistry.hmePrimerScript('PreToolUse');

const NATIVE_PRETOOL = nativeHooks.preToolHandlers;
const NATIVE_POSTTOOL = nativeHooks.postToolHandlers;
const OPENCODE_OBSERVATION_EVENTS = routeRegistry.observationEvents();

// Finish one bash hook invocation and return {stdout, stderr, exit_code}.
// Hook errors are recorded instead of thrown.
function _finishHook(eventName, scriptPath, startedAt, result) {
  const code = result.exit_code ?? 0;
  appendHookExec({
    event: eventName || 'hook',
    script: path.basename(scriptPath),
    cwd: process.cwd(),
    session_id: '',
    exit_code: code,
    duration_ms: Date.now() - startedAt,
    stdout_bytes: Buffer.byteLength(result.stdout || ''),
    stderr_bytes: Buffer.byteLength(result.stderr || ''),
  });
  if (code !== 0 && !_policyDecisionOutput(result.stdout || '')) {
    _appendHookFailure(eventName, scriptPath, code, result);
  }
  recordHookCheckpoint(PROJECT_ROOT, 'hook-script', {
    event: eventName || 'hook',
    policy: path.basename(scriptPath),
    decision: code === 0 ? 'ok' : 'error',
    exit_code: code,
    stdout_bytes: Buffer.byteLength(result.stdout || ''),
    stderr_bytes: Buffer.byteLength(result.stderr || ''),
  });
  return result;
}

function _policyDecisionOutput(stdout) {
  return /"decision"\s*:\s*"block"/.test(stdout)
    || /"ok"\s*:\s*false/.test(stdout)
    || /"permissionDecision"\s*:\s*"deny"/.test(stdout);
}

function _appendHookFailure(eventName, scriptPath, code, result) {
  try {
    const file = path.join(PROJECT_ROOT, 'log', 'hme-errors.log');
    const stderr = String(result.stderr || '').replace(/\s+/g, ' ').trim().slice(0, 400);
    const stdout = String(result.stdout || '').replace(/\s+/g, ' ').trim().slice(0, 200);
    const msg = `[${new Date().toISOString()}] [hook-failure] ${eventName}:${path.basename(scriptPath)} exit=${code}`
      + (stderr ? ` stderr=${stderr}` : '')
      + (stdout ? ` stdout=${stdout}` : '');
    fs.mkdirSync(path.dirname(file), { recursive: true });
    fs.appendFileSync(file, `${msg}\n`);
  } catch (_e) {
    // silent-ok: hook failure mirroring must never cause a second hook failure.
  }
}

function lifecycleContextResult(eventName, result) {
  const stderr = String(result && result.stderr || '').trim();
  if (!stderr || stderr === 'ok') return result;
  if (!new Set(['SessionStart', 'UserPromptSubmit', 'PreCompact', 'PostCompact']).has(eventName)) return result;
  if ((result.stdout || '').trim()) return result;
  return {
    stdout: renderInstruct(eventName, [stderr]),
    stderr: ' ',
    exit_code: result.exit_code || 0,
  };
}

function runHook(scriptPath, stdinJson, timeoutMs = 30_000, eventName = 'hook') {
  const startedAt = Date.now();
  return spawnFileInput('bash', [scriptPath], {
    input: stdinJson,
    timeoutMs,
    label: `${eventName}-${path.basename(scriptPath)}`,
    env: { PROJECT_ROOT, HME_HOOK_EVENT: eventName },
  }).then((result) => lifecycleContextResult(eventName, _finishHook(eventName, scriptPath, startedAt, {
    stdout: result.stdout,
    stderr: result.stderr,
    exit_code: result.exit_code,
  })));
}

// Run hook chains with shared stdin; outputs concatenate and first nonzero exit wins.
// Blocking decision JSON halts the chain because it supersedes later hooks.
async function runChain(scripts, stdinJson, timeoutMs = 30_000, eventName = 'hook') {
  let combinedStdout = '';
  let combinedStderr = '';
  let firstNonZeroCode = 0;
  for (const script of scripts) {
    const r = await runHook(script, stdinJson, timeoutMs, eventName);
    combinedStdout += r.stdout;
    combinedStderr += r.stderr;
    if (r.exit_code !== 0 && firstNonZeroCode === 0) firstNonZeroCode = r.exit_code;
    // Early-exit on block decision (stop/pretooluse hooks may emit JSON block).
    if (/\"decision\"\s*:\s*\"block\"/.test(r.stdout) || /\"ok\"\s*:\s*false/.test(r.stdout)) break;
  }
  // prevent Claude Code from displaying empty stderr as a hook error
  if (!combinedStderr && firstNonZeroCode === 0) combinedStderr = ' ';
  return { stdout: combinedStdout, stderr: combinedStderr, exit_code: firstNonZeroCode };
}

// Unified policy adapter returns runChain-shaped output or null when no policy fires.
// First-deny wins; later policies still run for side effects like stop_chain.
function _failClosedPolicyError(message, eventName) {
  return {
    stdout: renderPolicyFailure(message, eventName),
    stderr: `[unified-policies] ${message}\n`,
    exit_code: eventName === 'PreToolUse' || eventName === 'PermissionRequest' ? 0 : 2,
  };
}

async function _runUnifiedPolicies(policyEventName, toolName, stdinJson, outputEventName = policyEventName) {
  let registry, config;
  try {
    registry = require('../policies/registry');
    config = require('../policies/config');
  } catch (err) {
    return _failClosedPolicyError(`UNIFIED POLICY LOAD FAILURE: ${err.message}`, outputEventName);
  }
  try {
    registry.loadBuiltins();
    const cfg = config.get();
    if (cfg.customPoliciesPath) {
      const customPath = path.isAbsolute(cfg.customPoliciesPath)
        ? cfg.customPoliciesPath
        : path.join(PROJECT_ROOT, cfg.customPoliciesPath);
      registry.loadCustom(customPath);
    }
    const policies = registry.matchingFor(policyEventName, toolName, config);
    if (policies.length === 0) return null;
    let payload;
    try { payload = JSON.parse(stdinJson || '{}'); } catch (_e) { payload = {}; }
    const ctx = {
      toolInput: payload.tool_input || {},
      toolName: payload.tool_name || toolName,
      sessionId: payload.session_id || '',
      payload,
      deny: registry.deny,
      instruct: registry.instruct,
      allow: registry.allow,
      rewrite: registry.rewrite,
      params: {},
    };
    const aggregate = await registry.runChain(policies, ctx);
    let combinedStderr = '';
    for (const e of aggregate.errors || []) combinedStderr += `[unified-policies] ${e.policy}: ${e.error}\n`;
    if (aggregate.rewrites && aggregate.rewrites.length && (outputEventName === 'PreToolUse' || outputEventName === 'PermissionRequest')) {
      try {
        const { recordPolicyRewrite } = require('./hook_decision_log');
        recordPolicyRewrite(PROJECT_ROOT, payload, aggregate.rewrites);
      } catch (_e) { /* silent-ok: telemetry must never block */ }
    }
    if (aggregate.firstDeny && aggregate.firstDeny.policy) {
      try {
        const { recordPolicyDeny } = require('./hook_decision_log');
        recordPolicyDeny(PROJECT_ROOT, payload, aggregate.firstDeny.policy, aggregate.firstDeny.reason);
      } catch (_e) { /* silent-ok: telemetry must never block */ }
    }
    const stdout = renderPolicyAggregate(aggregate, { eventName: outputEventName, toolInput: ctx.toolInput });
    if (stdout) return { stdout, stderr: combinedStderr || ' ', exit_code: 0 };
    return null;
  } catch (err) {
    // silent-ok: failure is rendered into stdout/stderr and returned to caller.
    return {
      stdout: renderPolicyFailure(`UNIFIED POLICY RUNTIME FAILURE: ${err.message}`, outputEventName),
      stderr: `[unified-policies] crash: ${err.message}\n`,
      exit_code: outputEventName === 'PreToolUse' || outputEventName === 'PermissionRequest' ? 0 : 2,
    };
  }
}

function _opencodeTextCompleteDecision(stdinJson) {
  let payload;
  try { payload = JSON.parse(stdinJson || '{}'); } catch (_e) { payload = {}; }
  const text = String(payload.text || '');
  if (!text) return null;
  try {
    const { _stripSlop } = require('../proxy/sse_slop_rewriter');
    const slop = _stripSlop(text);
    if (slop && Array.isArray(slop.hits) && slop.hits.length > 0 && slop.out !== text) {
      return { stdout: JSON.stringify({ decision: slop.out ? 'modify' : 'drop', text: slop.out || '', reason: `HME slop rewrite: ${slop.hits.join(',')}` }), stderr: ' ', exit_code: 0 };
    }
  } catch (_e) { /* silent-ok: optional slop policy must not break hooks */ }
  try {
    const { evaluateStreamTextBlock } = require('../omo_bridge/stream_text_block_policy');
    const event = {
      abi: UNIVERSAL_HOOK_ABI,
      id: `hme-opencode-text-${payload.session_id || payload.sessionID || 'unknown'}`,
      timestamp: new Date().toISOString(),
      source: { host: 'opencode', adapter: 'dispatcher', rawEventName: 'TextComplete' },
      phase: 'stream.text_block',
      session: { id: payload.session_id || payload.sessionID || 'unknown', agent: 'opencode', provider: 'opencode' },
      stream: { text },
      payload,
    };
    const result = evaluateStreamTextBlock(event, { ctx: {}, slot: null });
    const decision = result && result.decision;
    if (!decision || decision.kind === 'allow') return null;
    if (decision.kind === 'drop') return { stdout: JSON.stringify({ decision: 'drop', reason: decision.reason || 'HME stream text block removed' }), stderr: ' ', exit_code: 0 };
    if (decision.kind === 'rewrite') return { stdout: JSON.stringify({ decision: 'modify', text: decision.text || '', reason: decision.reason || 'HME stream text block rewritten' }), stderr: ' ', exit_code: 0 };
  } catch (_e) { /* silent-ok: optional stream-text policy must not break hooks */ }
  return null;
}

// Parse tool_name from pre/post-tool payloads.
// Fall back to empty string on malformed JSON.
function _toolName(stdinJson) {
  try {
    const d = JSON.parse(stdinJson);
    return typeof d.tool_name === 'string' ? d.tool_name : '';
  } catch (_) { return ''; }
}

// Main entry point: dispatch eventName/stdinJson and return the forwarder response shape
// Host adapters relay this stdout/stderr/exit_code to their plugin protocol.
async function dispatchEvent(eventName, stdinJson) {
  const empty = stdinJson || '{}';
  if (shouldSkipForNestedHooks(eventName, empty)) return { stdout: '', stderr: ' ', exit_code: 0 };
  await _recordLifecycleState(eventName, empty);
  if (OPENCODE_OBSERVATION_EVENTS.has(eventName)) {
    if (eventName === 'TextComplete') {
      const streamDecision = _opencodeTextCompleteDecision(empty);
      if (streamDecision) return streamDecision;
    }
    const omo = await applyOmoLive(eventName, empty);
    if (omo.status === 'disabled') await observeOmoShadow(eventName, empty);
    return omo.applied && omo.result ? omo.result : { stdout: '', stderr: ' ', exit_code: 0 };
  }
  switch (eventName) {
    case 'SessionStart':
      if (!isStrictMode()) return { stdout: '', stderr: ' ', exit_code: 0 };
      await applyOmoLive('SessionStart', empty);
      await observeOmoShadow('SessionStart', empty);
      return runChain(routeRegistry.lifecycleScripts('SessionStart').map(_hookScript), empty, 30_000, 'SessionStart');
    case 'UserPromptSubmit':
      return runChain(routeRegistry.lifecycleScripts('UserPromptSubmit').map(_hookScript), empty, 30_000, 'UserPromptSubmit');
    case 'Stop': {
      const omo = await applyOmoLive('Stop', empty);
      if (omo.status === 'disabled') await observeOmoShadow('Stop', empty);
      // stop_chain evaluator: first-deny-wins, shell stages wrapped via shell_policy
      const stopChain = require('../proxy/stop_chain');
      const result = await stopChain.runStopChain(empty);
      // dominance rewriter removed -- was eating deny messages; re-add via enhance only
      if (result && /"decision"\s*:\s*"block"/.test(result.stdout || '')) return result;
      return omo.applied && omo.result ? omo.result : result;
    }
    case 'PreCompact':
      if (!isStrictMode()) return { stdout: '', stderr: ' ', exit_code: 0 };
      return runChain(routeRegistry.lifecycleScripts('PreCompact').map(_hookScript), empty, 30_000, 'PreCompact');
    case 'PostCompact':
      if (!isStrictMode()) return { stdout: '', stderr: ' ', exit_code: 0 };
      return runChain(routeRegistry.lifecycleScripts('PostCompact').map(_hookScript), empty, 30_000, 'PostCompact');
    case 'PreToolUse': {
      const omo = await applyOmoLive('PreToolUse', empty);
      if (omo.status === 'disabled') await observeOmoShadow('PreToolUse', empty);
      if (omo.applied && omo.result && /"permissionDecision"\s*:\s*"deny"/.test(omo.result.stdout || '')) return omo.result;
      const activeInput = omo.stdinJson || empty;
      const tool = _toolName(empty);
      const env = normalize(activeInput);
      const retry = _retryBlock(tool, env.tool_input || {});
      if (retry) return retry;
      if (isWriteFamilyTool(tool)) {
        const decision = await preWriteCheck(activeInput);
        const stdout = toHookResponse(decision);
        if (stdout || decision.permissionDecision !== 'allow') return { stdout, stderr: ' ', exit_code: 0 };
      }
      const unifiedRes = await _runUnifiedPolicies('PreToolUse', tool, activeInput);
      if (unifiedRes && unifiedRes.stdout) return unifiedRes;
      if (NATIVE_PRETOOL[tool]) {
        const native = await NATIVE_PRETOOL[tool](activeInput);
        if (native && native.stdout) return native;
        return omo.applied && omo.result ? omo.result : native;
      }
      const scripts = [...(PRETOOL_SCRIPTS[tool] || [])];
      // HME primer runs before first HME_* tool each session -- always chain it
      // for any HME_-prefixed tool, the primer self-guards against re-fire.
      if ((tool.startsWith('HME_') || tool.startsWith('mcp__HME__')) && HME_PRIMER_SCRIPT) {
        scripts.unshift(_hookScript(HME_PRIMER_SCRIPT));
      }
      if (scripts.length === 0) return omo.applied && omo.result ? omo.result : { stdout: '', stderr: ' ', exit_code: 0 };
      const chained = await runChain(scripts, activeInput, 30_000, 'PreToolUse');
      if (chained && chained.stdout) return chained;
      return omo.applied && omo.result ? omo.result : chained;
    }
    case 'PermissionRequest': {
      const omo = await applyOmoLive('PermissionRequest', empty);
      if (omo.status === 'disabled') await observeOmoShadow('PermissionRequest', empty);
      if (omo.applied && omo.result) return omo.result;
      const tool = _toolName(empty);
      // PermissionRequest reuses the PreToolUse policy context -- declared in
      // dispatcher-routes.json, not hardcoded, so the reuse is auditable.
      const unifiedRes = await _runUnifiedPolicies(policyContext('PermissionRequest'), tool, empty, 'PermissionRequest');
      return unifiedRes && unifiedRes.stdout ? unifiedRes : { stdout: '', stderr: ' ', exit_code: 0 };
    }
    case 'PostToolUse': {
      await _recordPostToolEvidence(empty);
      await observeOmoShadow('PostToolUse', empty);
      const tool = _toolName(empty);
      const env = normalize(empty);
      const response = env.tool_response || {};
      const exitCode = Number.isInteger(response.exit_code) ? response.exit_code : null;
      if (response && (response.is_error === true || response.error === true || (exitCode !== null && exitCode !== 0))) _recordToolFailure(tool, env.tool_input || {}, response);
      const unifiedRes = await _runUnifiedPolicies('PostToolUse', tool, empty);
      if (unifiedRes && unifiedRes.stdout) return unifiedRes;
      if (NATIVE_POSTTOOL[tool]) {
        const scripts = [...UNIVERSAL_POSTTOOL, ...(POSTTOOL_SCRIPTS[tool] || [])];
        const universal = await runChain(scripts, empty, 30_000, 'PostToolUse');
        const native = await NATIVE_POSTTOOL[tool](empty);
        return {
          stdout: `${universal.stdout || ''}${native.stdout || ''}`,
          stderr: `${universal.stderr || ''}${native.stderr || ''}` || ' ',
          exit_code: universal.exit_code || native.exit_code || 0,
        };
      }
      const scripts = [...UNIVERSAL_POSTTOOL, ...(POSTTOOL_SCRIPTS[tool] || [])];
      return runChain(scripts, empty, 30_000, 'PostToolUse');
    }
    default:
      return {
        stdout: '',
        stderr: `[event_kernel] unknown event: ${eventName}`,
        exit_code: 0,
      };
  }
}

module.exports = { dispatchEvent, runHook, runChain, lifecycleContextResult, policyContext };