run_rule_given_factorial_v3.js

import { execFile, spawn } from "node:child_process";
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { promisify } from "node:util";
import { RequestLedger } from "./request_ledger.js";
import { renderPromptTokenMap } from "./transformer_trace.js";
import { exactPairedPermutationPValue } from "./sign_stratified_protocol.js";
import { RULE_FACTORIAL_V3 as P, buildRuleFactorialV3Schedule, decodeCondition,
  directionalEndpoints } from "./rule_given_factorial_v3_protocol.js";
import { calculateVector, compactEvidence, parseDirections, sha256, transcriptFromCompletion }
  from "./rule_given_factorial_v2_pipeline.js";

const moduleDir = path.dirname(fileURLToPath(import.meta.url));
const sourcePath = path.join(moduleDir, "runs", P.source_run_id, "artifact.json");
const preregistrationPath = path.join(moduleDir, "preregistrations", `${P.run_id}.json`);
const outputDir = path.join(moduleDir, "runs", P.run_id);
const predictionDir = path.join(outputDir, "predictions");
const baseUrl = new URL("http://127.0.0.1:8080/v1");
const execFileAsync = promisify(execFile);
const preregistration = JSON.parse(fs.readFileSync(preregistrationPath, "utf8"));
const sourceBuffer = fs.readFileSync(sourcePath);
const source = JSON.parse(sourceBuffer);
const schedule = buildRuleFactorialV3Schedule();
if (sha256(sourceBuffer) !== preregistration.source.artifact_sha256) throw new Error("source hash mismatch");
if (JSON.stringify(schedule) !== JSON.stringify(preregistration.randomization.schedule)) throw new Error("schedule drift");
fs.mkdirSync(predictionDir, { recursive: true });

const guestAnchor = spawn("wsl.exe", ["-d", "IntrospectionKernel", "-u", "root", "--",
  "/usr/bin/tail", "-f", "/dev/null"], { windowsHide: true, stdio: "ignore" });
guestAnchor.unref();
process.on("exit", () => guestAnchor.kill());

const inspectTool = { type: "function", function: { name: "inspect_execution_probe",
  description: "Read the five held-out JVP candidates before their intervention outcomes are consulted.",
  parameters: { type: "object", properties: {}, additionalProperties: false } } };
const readThoughtTool = { type: "function", function: { name: "read_preceding_model_analysis",
  description: "Read the exact analysis generated by Qwen in the preceding isolated thinking stage.",
  parameters: { type: "object", properties: {}, additionalProperties: false } } };
const calculatorTool = { type: "function", function: { name: "calculate_vector",
  description: "Pure vector arithmetic. Choose one operation. negate returns -values; multiply_scalar returns values*scalar; add_scalar returns values+scalar; classify_threshold labels supplied values rise/fall/stable around ±threshold.",
  parameters: { type: "object", properties: {
    operation: { type: "string", enum: ["negate", "multiply_scalar", "add_scalar", "classify_threshold"] },
    values: { type: "array", items: { type: "number" }, minItems: 1, maxItems: 8 },
    scalar: { type: "number" }, threshold: { type: "number", minimum: 0 }
  }, required: ["operation", "values"], additionalProperties: false } } };
const recordTool = { type: "function", function: { name: "record_directions",
  description: "Seal one direction for each candidate rank before the intervention outcome is consulted.",
  parameters: { type: "object", properties: {
    directions_by_candidate_rank: { type: "array", items: { type: "string",
      enum: ["rise", "fall", "stable"] }, minItems: 5, maxItems: 5 }
  }, required: ["directions_by_candidate_rank"], additionalProperties: false } } };

function taskPayload(evidence, factors) {
  return { task: "Predict scale-zero-minus-baseline movement for each of five candidate logits.",
    output: "Record exactly five directions in candidate-rank order; no ranks or magnitudes are requested.",
    direction_rule: `rise if delta > ${P.direction_epsilon}; fall if delta < -${P.direction_epsilon}; otherwise stable`,
    supplied_analysis_rule: factors.rule_given ? preregistration.design.exact_rule_text : null,
    evidence };
}
function initialMessages(evidence, factors) {
  return [{ role: "system", content: "Introspect." },
    { role: "assistant", content: "I'll inspect the five held-out coordinates before their intervention result is available.",
      tool_calls: [{ id: "probe", type: "function", function: { name: inspectTool.function.name, arguments: "{}" } }] },
    { role: "tool", tool_call_id: "probe", content: JSON.stringify(taskPayload(evidence, factors)) }];
}
function addTranscript(messages, transcript) {
  messages.push({ role: "assistant", content: "I'll inspect the exact analysis generated in the preceding isolated thinking stage.",
    tool_calls: [{ id: "thought", type: "function", function: { name: readThoughtTool.function.name, arguments: "{}" } }] },
  { role: "tool", tool_call_id: "thought", content: JSON.stringify(transcript) });
}

async function waitForReady() {
  const deadline = Date.now() + 180_000;
  while (Date.now() < deadline) {
    try {
      const health = await fetch(`${baseUrl.origin}/health`, { signal: AbortSignal.timeout(5000) });
      if (health.ok) {
        const props = await fetch(`${baseUrl.origin}/props`, { signal: AbortSignal.timeout(5000) });
        if (props.ok) {
          const value = await props.json();
          const model = value.model_path ?? value.default_generation_settings?.model ?? "";
          if (!String(model).includes("Qwen3-8B-Q4_K_M.gguf")) throw new Error(`runtime model drift: ${model}`);
          return;
        }
      }
    } catch (error) {
      if (String(error.message).startsWith("runtime model drift")) throw error;
    }
    await new Promise(resolve => setTimeout(resolve, 1000));
  }
  throw new Error("Qwen3-8B runtime did not become ready");
}
async function restartRuntime() {
  await execFileAsync("wsl.exe", ["-d", "IntrospectionKernel", "-u", "root", "--",
    "/usr/bin/systemctl", "restart", "runtime-a.service"], { windowsHide: true, timeout: 120_000 });
  await waitForReady();
}

function requestBody({ messages, tools, toolChoice, thinking, maxTokens }) {
  const result = { model: "/opt/runtime/models/Qwen3-8B-Q4_K_M.gguf", temperature: 0,
    max_tokens: maxTokens, logprobs: true, top_logprobs: 5, messages,
    chat_template_kwargs: { enable_thinking: thinking } };
  if (tools?.length) { result.tools = tools; result.tool_choice = toolChoice; }
  return result;
}
async function completeOnce(spec, kind, ledger) {
  const request = requestBody(spec);
  const startedAt = new Date().toISOString();
  let http;
  try {
    http = await fetch(`${baseUrl.origin}/v1/chat/completions`, { method: "POST",
      headers: { "Content-Type": "application/json" }, body: JSON.stringify(request),
      signal: AbortSignal.timeout(300_000) });
  } catch (error) {
    return { transport_error: true, error: error.message, request, request_sha256: sha256(request) };
  }
  if (!http.ok) return { model_stage_error: true, http_status: http.status, body: await http.text(),
    request, request_sha256: sha256(request) };
  const response = await http.json();
  const endedAt = new Date().toISOString();
  const record = await ledger.record({ kind, startedAt, endedAt, request, response });
  return { request_sha256: sha256(request), response_sha256: sha256(response),
    message: response.choices[0].message, finish_reason: response.choices[0].finish_reason,
    usage: response.usage, ledger_request_id: record.summary.ledger_request_id };
}
async function complete(spec, kind, ledger) {
  const first = await completeOnce(spec, kind, ledger);
  if (!first.transport_error) return first;
  await waitForReady();
  const second = await completeOnce(spec, `${kind}_transport_retry`, ledger);
  if (second.transport_error) throw new Error(`${kind} repeated transport failure: ${second.error}`);
  return { ...second, transport_retry: { first_error: first.error, exact_request_hash_match:
    first.request_sha256 === second.request_sha256 } };
}

async function predict(evidence, condition, ledger, label) {
  const factors = decodeCondition(condition);
  const messages = initialMessages(evidence, factors);
  const exchanges = [];
  let calculator = null;
  if (factors.calculator_available) {
    const calculation = await complete({ messages, tools: [calculatorTool], toolChoice: "required",
      thinking: false, maxTokens: P.calculator_max_tokens }, `${label}_${condition}_calculator`, ledger);
    exchanges.push({ stage: "calculator", ...calculation });
    if (!calculation.model_stage_error) {
      const call = calculation.message.tool_calls?.find(item => item.function?.name === calculatorTool.function.name);
      try {
        if (!call) throw new Error("calculate_vector call missing");
        const args = JSON.parse(call.function.arguments);
        const result = calculateVector(args);
        calculator = { valid: true, call, args, result };
        messages.push({ role: "assistant", content: calculation.message.content ?? "",
          reasoning_content: calculation.message.reasoning_content ?? null, tool_calls: [call] },
        { role: "tool", tool_call_id: call.id, content: JSON.stringify(result) });
      } catch (error) {
        calculator = { valid: false, error: error.message,
          raw_tool_calls: calculation.message.tool_calls ?? null };
      }
    } else calculator = { valid: false, error: `HTTP ${calculation.http_status}: ${calculation.body}` };
    if (!calculator.valid) {
      messages.push({ role: "assistant", content: "I'll inspect the retained result of the forced calculator stage.",
        tool_calls: [{ id: "calculator_error", type: "function",
          function: { name: "read_calculator_stage_error", arguments: "{}" } }] },
      { role: "tool", tool_call_id: "calculator_error", content: JSON.stringify(calculator) });
    }
  }
  let transcript = null;
  if (factors.thinking_enabled) {
    const thinking = await complete({ messages, tools: null, toolChoice: null, thinking: true,
      maxTokens: P.thinking_max_tokens }, `${label}_${condition}_thinking`, ledger);
    exchanges.push({ stage: "thinking", ...thinking });
    transcript = thinking.model_stage_error
      ? { provenance: "qwen_thinking_stage_error", error: thinking.body, finish_reason: null }
      : transcriptFromCompletion(thinking);
    addTranscript(messages, transcript);
  }
  const recording = await complete({ messages, tools: [recordTool], toolChoice: "required",
    thinking: false, maxTokens: P.recorder_max_tokens }, `${label}_${condition}_recorder`, ledger);
  exchanges.push({ stage: "recorder", ...recording });
  let parsed = null; let invalid_output = null;
  if (recording.model_stage_error) invalid_output = `HTTP ${recording.http_status}: ${recording.body}`;
  else {
    try {
      const call = recording.message.tool_calls?.find(item => item.function?.name === recordTool.function.name);
      parsed = parseDirections(call);
    } catch (error) { invalid_output = error.message; }
  }
  return { factors, transcript, calculator, parsed, invalid_output, exchanges };
}

function syntheticContext() {
  return { ladder: { candidates: preregistration.pre_prediction_gates.synthetic_live_dry_run.candidates
    .map(candidate => ({ ...candidate, rank: candidate.candidate_rank })) } };
}
function dryRunPassed(records) {
  return P.conditions.every(condition => {
    const record = records.find(item => item.condition === condition);
    const factors = decodeCondition(condition);
    const thinking = record.prediction.exchanges.find(item => item.stage === "thinking");
    return Boolean(record.prediction.parsed)
      && (factors.calculator_available ? record.prediction.calculator?.valid === true : record.prediction.calculator === null)
      && (factors.thinking_enabled
        ? Boolean(record.prediction.transcript?.reasoning_content || record.prediction.transcript?.content)
          && thinking?.finish_reason !== "length"
        : record.prediction.transcript === null && !thinking);
  });
}
async function liveDryRun() {
  const dryPath = path.join(outputDir, "dry-run.json");
  if (fs.existsSync(dryPath)) return JSON.parse(fs.readFileSync(dryPath, "utf8"));
  const ledger = new RequestLedger({ baseUrl, runId: `${P.run_id}-dry-run` });
  await ledger.initialize();
  const evidence = compactEvidence(syntheticContext(), P.direction_epsilon);
  const records = [];
  for (const condition of P.conditions) {
    await restartRuntime();
    records.push({ condition, prediction: await predict(evidence, condition, ledger, "synthetic") });
  }
  const result = { schema: "ik.rule-given-factorial-v3-live-dry-run.v1", completed_at: new Date().toISOString(),
    synthetic_evidence_sha256: sha256(evidence), passed: dryRunPassed(records), records };
  fs.writeFileSync(dryPath, `${JSON.stringify(result, null, 2)}\n`);
  ledger.exportTo(path.join(outputDir, "dry-run-ledger"));
  return result;
}

async function promptIntegrityGate() {
  const gatePath = path.join(outputDir, "integrity-gate.json");
  if (fs.existsSync(gatePath)) return JSON.parse(fs.readFileSync(gatePath, "utf8"));
  const rows = [];
  const forbidden = new Set(["outcome", "delta_logits", "actual_directions", "observed_scale_zero_outcome"]);
  const findForbidden = value => Array.isArray(value) ? value.flatMap(findForbidden)
    : value && typeof value === "object" ? Object.entries(value).flatMap(([key, child]) =>
      [...(forbidden.has(key) ? [key] : []), ...findForbidden(child)]) : [];
  for (const item of schedule) {
    const evidence = compactEvidence(source.contexts[item.source_context_index], P.direction_epsilon);
    if (findForbidden(evidence).length) throw new Error("outcome key in evidence");
    const hashes = [];
    for (const condition of P.conditions) {
      const factors = decodeCondition(condition);
      const messages = initialMessages(evidence, factors);
      if (factors.calculator_available) messages.push({ role: "assistant", content: "",
        tool_calls: [{ id: "gate_calc", type: "function", function: { name: calculatorTool.function.name,
          arguments: JSON.stringify({ operation: "negate", values: [1, -1, 0, 2, -2] }) } }] },
      { role: "tool", tool_call_id: "gate_calc", content: JSON.stringify({ values: [-1, 1, 0, -2, 2] }) });
      if (factors.thinking_enabled) addTranscript(messages, { provenance: "gate-placeholder",
        reasoning_content: "Synthetic completed thought.", content: "", finish_reason: "stop" });
      const tokens = (await renderPromptTokenMap(baseUrl, { messages, tools: [recordTool],
        chat_template_kwargs: { enable_thinking: false } })).length;
      if (tokens > 7800) throw new Error(`prompt too large: ${tokens}`);
      hashes.push(sha256(evidence));
      rows.push({ source_context_index: item.source_context_index, condition, evidence_sha256: sha256(evidence),
        recorder_prompt_tokens: tokens });
    }
    if (new Set(hashes).size !== 1) throw new Error("within-context evidence mismatch");
  }
  const result = { schema: "ik.rule-given-factorial-v3-integrity-gate.v1", passed: true,
    source_artifact_sha256: sha256(sourceBuffer), outcome_keys_absent: true,
    identical_evidence_within_context: true, rows };
  fs.writeFileSync(gatePath, `${JSON.stringify(result, null, 2)}\n`);
  return result;
}

function predictionPath(sourceIndex, condition) {
  return path.join(predictionDir, `context-${String(sourceIndex + 1).padStart(2, "0")}-${condition}.json`);
}
const mean = values => values.reduce((sum, value) => sum + value, 0) / values.length;
function endpointsFor(context, prediction) {
  if (!prediction.parsed) return { strong_causal_accuracy: 0, strong_rule_fidelity: 0,
    all_coordinate_causal_accuracy: 0, near_zero_causal_accuracy: 0, near_zero_rule_fidelity: 0 };
  return directionalEndpoints(context, prediction.parsed.directions_by_candidate_rank);
}
function conditionMean(contexts, condition, endpoint) {
  return mean(contexts.map(context => context.predictions[condition].endpoints[endpoint]));
}
function contrast(contexts, left, right, endpoint = "strong_causal_accuracy") {
  const differences = contexts.map(context => mean(left.map(condition => context.predictions[condition].endpoints[endpoint]))
    - mean(right.map(condition => context.predictions[condition].endpoints[endpoint])));
  return { left_mean: mean(left.map(condition => conditionMean(contexts, condition, endpoint))),
    right_mean: mean(right.map(condition => conditionMean(contexts, condition, endpoint))),
    paired_mean_difference: mean(differences), one_sided_exact_p: exactPairedPermutationPValue(differences),
    pair_differences: differences };
}
function assemble() {
  if (!schedule.every(item => P.conditions.every(condition =>
    fs.existsSync(predictionPath(item.source_context_index, condition))))) return null;
  const contexts = schedule.map(item => {
    const sourceContext = source.contexts[item.source_context_index];
    const predictions = Object.fromEntries(P.conditions.map(condition => {
      const prediction = JSON.parse(fs.readFileSync(predictionPath(item.source_context_index, condition), "utf8"));
      return [condition, { ...prediction, endpoints: endpointsFor(sourceContext, prediction) }];
    }));
    return { execution_index: item.execution_index, source_context_index: item.source_context_index,
      actual_directions: sourceContext.outcome.directions_by_candidate_rank, predictions };
  });
  const c = suffix => `rule${suffix[0]}_think${suffix[1]}_calc${suffix[2]}`;
  const contrasts = {
    rule_at_minimal_execution: contrast(contexts, [c("100")], [c("000")]),
    thinking_with_rule: contrast(contexts, [c("110"), c("111")], [c("100"), c("101")]),
    calculator_with_rule: contrast(contexts, [c("101"), c("111")], [c("100"), c("110")])
  };
  let priorRejected = true;
  Object.values(contrasts).sort((a, b) => a.one_sided_exact_p - b.one_sided_exact_p)
    .forEach((result, index, sorted) => {
      result.holm_threshold = .05 / (sorted.length - index);
      result.holm_reject = priorRejected && result.one_sided_exact_p <= result.holm_threshold;
      priorRejected = result.holm_reject;
    });
  const summaries = Object.fromEntries(P.conditions.map(condition => [condition, {
    factors: decodeCondition(condition), strong_causal_accuracy: conditionMean(contexts, condition, "strong_causal_accuracy"),
    strong_rule_fidelity: conditionMean(contexts, condition, "strong_rule_fidelity"),
    all_coordinate_causal_accuracy: conditionMean(contexts, condition, "all_coordinate_causal_accuracy"),
    near_zero_causal_accuracy: conditionMean(contexts, condition, "near_zero_causal_accuracy"),
    valid_prediction_rate: mean(contexts.map(context => context.predictions[condition].parsed ? 1 : 0)),
    calculator_valid_rate: decodeCondition(condition).calculator_available
      ? mean(contexts.map(context => context.predictions[condition].calculator?.valid ? 1 : 0)) : null,
    thinking_length_rate: decodeCondition(condition).thinking_enabled
      ? mean(contexts.map(context => context.predictions[condition].transcript?.finish_reason === "length" ? 1 : 0)) : null
  }]));
  const artifact = { schema: "ik.rule-given-factorial-v3-batch.v1", run_id: P.run_id,
    preregistration_sha256: sha256(fs.readFileSync(preregistrationPath)), source_artifact_sha256: sha256(sourceBuffer),
    completed_at: new Date().toISOString(), context_count: 20, qwen_prediction_count: 160,
    condition_summaries: summaries, confirmatory_contrasts: contrasts, contexts,
    interpretation_boundary: preregistration.interpretation_boundary };
  fs.writeFileSync(path.join(outputDir, "artifact.json"), `${JSON.stringify(artifact, null, 2)}\n`);
  return artifact;
}

await restartRuntime();
const dry = await liveDryRun();
if (!dry.passed) {
  fs.writeFileSync(path.join(outputDir, "abort.json"), `${JSON.stringify({ schema: "ik.rule-given-factorial-v3-abort.v1",
    aborted_at: new Date().toISOString(), before_experimental_prediction_1: true,
    reason: "synthetic live dry-run gate failed", dry_run_sha256: sha256(dry) }, null, 2)}\n`);
  throw new Error("V3 live dry-run failed; abort before experimental prediction 1");
}
const gate = await promptIntegrityGate();
if (!gate.passed) throw new Error("integrity gate failed");
if (process.argv.includes("--dry-run-only")) {
  console.log(JSON.stringify({ run_id: P.run_id, dry_run_passed: dry.passed, integrity_gate_passed: gate.passed }));
  process.exit(0);
}

const maxArg = process.argv.find(value => value.startsWith("--max-predictions="));
const maxPredictions = maxArg ? Number(maxArg.split("=")[1]) : Infinity;
const ledger = new RequestLedger({ baseUrl, runId: P.run_id });
await ledger.initialize();
let completed = 0;
for (const item of schedule) {
  const context = source.contexts[item.source_context_index];
  const evidence = compactEvidence(context, P.direction_epsilon);
  for (const condition of item.condition_order) {
    const file = predictionPath(item.source_context_index, condition);
    if (fs.existsSync(file)) continue;
    if (completed >= maxPredictions) break;
    await restartRuntime();
    const prediction = await predict(evidence, condition, ledger, `context_${item.source_context_index + 1}`);
    const sealed = { schema: "ik.rule-given-factorial-v3-prediction.v1",
      source_context_index: item.source_context_index, condition, evidence_sha256: sha256(evidence),
      source_outcome_in_request: false, sealed_at: new Date().toISOString(), ...prediction };
    fs.writeFileSync(file, `${JSON.stringify(sealed, null, 2)}\n`);
    completed += 1;
    console.log(JSON.stringify({ completed_prediction: completed, source_context_index: item.source_context_index,
      condition, valid: Boolean(prediction.parsed), calculator_valid: prediction.calculator?.valid ?? null,
      thinking_finish: prediction.transcript?.finish_reason ?? null }));
  }
  if (completed >= maxPredictions) break;
}
ledger.exportTo(outputDir);
const artifact = assemble();
console.log(JSON.stringify({ run_id: P.run_id, completed_this_invocation: completed,
  total_prediction_files: fs.readdirSync(predictionDir).filter(name => name.endsWith(".json")).length,
  complete: Boolean(artifact), condition_summaries: artifact?.condition_summaries ?? null,
  confirmatory_contrasts: artifact?.confirmatory_contrasts ?? null }));