上流タスクが今日ちゃんと動いたかを、下流タスクが自分で確かめる — 無人スケジューラの完了台帳と依存バリア

// record.ts
import { createHash } from "node:crypto";
import { readFile, writeFile, rename, mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { RunRecord } from "./ledger";
 
export function fingerprintOf(content: string): string {
  return createHash("sha256").update(content).digest("hex").slice(0, 16);
}
 
// JST の YYYY-MM-DD を明示的に生成する（UTC ずれ防止）
export function jstDate(now = new Date()): string {
  const jst = new Date(now.getTime() + 9 * 60 * 60 * 1000);
  return jst.toISOString().slice(0, 10);
}
 
export async function recordRun(
  ledgerPath: string,
  rec: RunRecord
): Promise<void> {
  let ledger: RunRecord[] = [];
  try {
    ledger = JSON.parse(await readFile(ledgerPath, "utf8"));
  } catch {
    ledger = []; // 初回、または壊れていた場合は作り直す
  }
 
  // 同じ task+date は最新で置き換える（再実行に対して冪等）
  const key = (r: RunRecord) => `${r.task}::${r.date}`;
  ledger = ledger.filter((r) => key(r) !== key(rec));
  ledger.push(rec);
 
  // 台帳が無限に伸びないよう直近 60 日だけ残す
  const cutoff = jstDate(new Date(Date.now() - 60 * 864e5));
  ledger = ledger.filter((r) => r.date >= cutoff);
 
  // 一時ファイル → rename の原子的差し替え
  const dir = await mkdtemp(join(tmpdir(), "ledger-"));
  const tmp = join(dir, "ledger.json");
  await writeFile(tmp, JSON.stringify(ledger, null, 2), "utf8");
  await rename(tmp, ledgerPath);
}

// barrier.ts
import { readFile } from "node:fs/promises";
import type { RunRecord } from "./ledger";
import { jstDate, fingerprintOf } from "./record";
 
export type Gate =
  | { decision: "proceed"; upstream: RunRecord }
  | { decision: "skip"; reason: string }
  | { decision: "halt"; reason: string };
 
export async function checkUpstream(
  ledgerPath: string,
  upstreamTask: string,
  opts: { expectArtifact?: string } = {}
): Promise<Gate> {
  const today = jstDate();
  let ledger: RunRecord[] = [];
  try {
    ledger = JSON.parse(await readFile(ledgerPath, "utf8"));
  } catch {
    return { decision: "halt", reason: "台帳が読めません（破損 or 未生成）" };
  }
 
  const rec = ledger.find((r) => r.task === upstreamTask && r.date === today);
 
  // 1) 未実行 — 前日の残骸で走らないよう、止まる
  if (!rec) {
    return { decision: "halt", reason: `${upstreamTask} が今日まだ完了していません` };
  }
  // 2) 失敗 — 上流の成果物は信用できない、止まる
  if (rec.status === "error") {
    return { decision: "halt", reason: `${upstreamTask} が失敗しています: ${rec.note ?? ""}` };
  }
  // 3) 正当な空 — エラーではない、静かにスキップ
  if (rec.status === "empty") {
    return { decision: "skip", reason: `${upstreamTask} は今日正当に空でした` };
  }
 
  // 4) 産出 — 下流が実際に読むファイルが、上流が記録した内容と一致するか照合
  if (opts.expectArtifact && rec.fingerprint) {
    const actual = fingerprintOf(await readFile(opts.expectArtifact, "utf8"));
    if (actual !== rec.fingerprint) {
      return {
        decision: "halt",
        reason: `成果物のハッシュ不一致（stale clone か部分書き込みの疑い）`,
      };
    }
  }
  return { decision: "proceed", upstream: rec };
}

// downstream.ts
import { checkUpstream } from "./barrier";
 
const LEDGER = "/data/pipeline/ledger.json";
 
async function main() {
  const gate = await checkUpstream(LEDGER, "daily-reference-and-ticker", {
    expectArtifact: "/data/reference/claudelab.md",
  });
 
  if (gate.decision === "halt") {
    console.error(`[BARRIER] 中止: ${gate.reason}`);
    process.exit(1); // スケジューラのログに失敗として残す
  }
  if (gate.decision === "skip") {
    console.log(`[BARRIER] スキップ: ${gate.reason}`);
    return; // 正常終了。成果物は作らない
  }
 
  // ここに来たときだけ、上流の鮮度が保証されている
  console.log(`[BARRIER] 続行: 上流 ${gate.upstream.finishedAt} 完了`);
  await generateArticleFromReference();
}
 
async function generateArticleFromReference() {
  // 実際の生成処理
}
 
main().catch((e) => {
  console.error(`[FATAL] ${e.message}`);
  process.exit(1);
});