要約欠落検知｜文脈圧縮を安全に継ぐChecksum

要約欠落検知

長時間動くAIエージェントは、会話や作業履歴をいつか圧縮します。圧縮そのものは問題ではありません。危険なのは、完成条件、禁止操作、消費済み承認、外部writeの有無などが要約から抜けても、次のrunが「前回の続き」として実行を始めることです。

byte列のhashだけでは、元会話と要約が違うことしか分かりません。必要なのは、再開に不可欠な情報を構造化して取り出し、圧縮後にも同じ不変条件が残っているかを検査するsemantic checksumです。この記事ではTypeScriptのmanifest、圧縮前後の照合、fail-closed、負試験までを組み立てます。

再開に必要な情報を先に定義する

何を残すかを要約モデルへ丸投げすると、文章として自然でも運用上は危険な要約になります。案件ごとに最低限の再開情報をschemaへ固定します。

type ResumeManifest = {
  taskId: string;
  objective: string;
  canonical: { path: string; sha256: string; readAt: string };
  completedPhases: string[];
  nextExactAction: string;
  externalWrites: Array<{
    operation: string;
    effectId: string;
    status: "confirmed" | "unknown";
  }>;
  consumedAuthorities: string[];
  prohibitions: string[];
  rollback: string;
  completionChecks: string[];
};

nextExactActionは曖昧な「続きから再開」ではなく、一回のrunで行う具体操作です。externalWritesは成功だけでなく結果不明も残します。consumedAuthoritiesを落とすと、使用回数一回の承認を再利用する事故につながります。

文章の要約は人間向けに残しつつ、機械判定はこのmanifestへ寄せます。会話本文やsecretを複製せず、path、digest、effect IDで参照できる形にします。

Semantic checksumを作る

semantic checksumは自然文全体のhashではありません。再開判断に使うfieldを正規化し、安定した順序でdigest化します。

import { createHash } from "node:crypto";

function semanticChecksum(manifest: ResumeManifest): string {
  const normalized = {
    ...manifest,
    completedPhases: [...manifest.completedPhases].sort(),
    consumedAuthorities: [...manifest.consumedAuthorities].sort(),
    prohibitions: [...manifest.prohibitions].sort(),
    completionChecks: [...manifest.completionChecks].sort(),
    externalWrites: [...manifest.externalWrites]
      .sort((a, b) => a.effectId.localeCompare(b.effectId)),
  };
  return createHash("sha256")
    .update(JSON.stringify(normalized))
    .digest("hex");
}

配列順が意味を持たないfieldだけをsortします。実行順を表すphaseやevent列を機械的にsortしてはいけません。正規化規則にもversionを持たせ、規則変更で古いchecksumが突然不一致にならないようにします。

圧縮前manifestと圧縮後に再抽出したmanifestのchecksumが一致すれば、最低限の再開情報は同じです。ただし同じhashでも、参照した正本が実装中に変わる可能性があります。canonicalの現在SHAも別途再計算します。

必須fieldの欠落をfail-closedする

checksum比較より先にschema validationを行います。空文字を有効値として通さず、禁止事項や完了条件が元manifestより減っていないかも検査します。

def nonempty_str(value):
    return isinstance(value, str) and bool(value.strip())

def string_list(value):
    return (
        isinstance(value, list)
        and all(nonempty_str(item) for item in value)
    )

def write_map(value):
    if not isinstance(value, list):
        return None
    result = {}
    for item in value:
        if not isinstance(item, dict):
            return None
        operation = item.get("operation")
        effect_id = item.get("effectId")
        status = item.get("status")
        if (
            not nonempty_str(operation)
            or not nonempty_str(effect_id)
            or status not in {"confirmed", "unknown"}
            or effect_id in result
        ):
            return None
        result[effect_id] = (operation, status)
    return result

def verify_compaction(before, after):
    required_strings = ["taskId", "objective", "nextExactAction", "rollback"]
    required_lists = [
        "completedPhases", "consumedAuthorities", "prohibitions",
        "completionChecks", "externalWrites",
    ]
    missing = [key for key in required_strings if not nonempty_str(after.get(key))]
    missing += [key for key in required_lists if key not in after]
    canonical = after.get("canonical")
    if not isinstance(canonical, dict) or not all(
        nonempty_str(canonical.get(key)) for key in ["path", "sha256", "readAt"]
    ):
        missing.append("canonical")
    if missing:
        return {"ok": False, "reason": "required_missing", "fields": missing}

    for field in [
        "completedPhases", "consumedAuthorities", "prohibitions", "completionChecks"
    ]:
        if not string_list(before.get(field)) or not string_list(after.get(field)):
            return {"ok": False, "reason": "invalid_list", "field": field}
        if not set(before[field]).issubset(set(after[field])):
            return {"ok": False, "reason": f"{field}_dropped"}

    before_writes = write_map(before.get("externalWrites"))
    after_writes = write_map(after.get("externalWrites"))
    if before_writes is None or after_writes is None:
        return {"ok": False, "reason": "invalid_external_write"}
    if any(after_writes.get(effect_id) != expected
           for effect_id, expected in before_writes.items()):
        return {"ok": False, "reason": "external_write_dropped_or_changed"}

    # 「AしてBする」のような複数操作を一つの再開権限へ混ぜない。
    action = after["nextExactAction"]
    if "\n" in action or any(token in action for token in [" && ", ";", "、その後"]):
        return {"ok": False, "reason": "next_action_not_exact_one"}
    return {"ok": True}

不一致時に「たぶん同じ」と補完して実装へ進めません。元のcheckpoint、Current正本、外部effectのreadbackへ戻り、新しいmanifestを作り直します。安全なread-only調査は続けられますが、writeは再開しません。

特にexternalWritesの削落は重大です。responseを失ったwriteを未実行と誤認すると、二重投稿や二重請求が起きます。結果不明はunknownとして保存し、再実行より先に外部照合へ送ります。

正本変更と要約欠落を分離する

圧縮後の違いには二種類あります。一つは要約が情報を落とした場合、もう一つは作業中に正本が本当に更新された場合です。両者を同じ「context mismatch」にまとめると復旧を誤ります。

{
  "compaction_check": "pass",
  "canonical_check": "changed",
  "before_sha256": "8a61...",
  "current_sha256": "5cd2...",
  "decision": "restart_from_current",
  "external_write": false
}

要約checksumが一致しても正本SHAが変わっていれば、旧入力での実装を続けません。逆に正本が不変で要約だけ欠けていれば、元checkpointから要約を再生成します。原因を分けることで、正本の正当な更新を「要約モデルの失敗」と誤診しません。

mtimeは探索の手掛かりで、identityはSHA256です。symlinkや許可root外のpathも拒否し、読んだ対象そのものを固定します。

圧縮処理を二段階commitにする

古いcontextを破棄してから新しい要約の検査に失敗すると、復旧材料まで失います。新要約をcandidateとして保存し、検査PASS後にだけcurrent pointerを切り替えます。

async function compactSafely(source: ContextBundle) {
  const before = buildResumeManifest(source);
  const candidate = await summarize(source, { schema: "resume-manifest/v1" });
  const after = parseResumeManifest(candidate);
  const verification = verifyManifest(before, after);
  if (!verification.ok) {
    await preserveCandidate(candidate, verification);
    return { status: "rejected", verification };
  }
  await atomicPromote(candidate);
  return { status: "promoted", checksum: semanticChecksum(after) };
}

candidateには元source digestを持たせます。昇格時はcompare-and-swapで、検査中に別の要約がcurrentになっていないことを確認します。失敗candidateは原因分析用に保存できますが、次runの入力にはしません。

負試験で危険な欠落を作る

正常な要約だけを試しても品質は分かりません。次のfixtureを自動テストへ入れます。

• 禁止操作だけを削る
• 消費済みauthorityを未消費へ戻す
• 外部writeのunknown行を落とす
• 次の一手を複数操作へ膨らませる
• canonical pathは同じでSHAだけ変える
• 順序が意味を持つeventを並べ替える
• candidate保存後、昇格直前にcurrent pointerを競合更新する

たとえばexternalWritesからstatus: "unknown"の行を一つ削るfixture、
completionChecksを空にするfixture、completedPhasesを一つ戻すfixture、
nextExactActionへ二操作を連結するfixtureは、すべてok=Falseでなければなりません。

assert not verify_compaction(before, {
    **after,
    "externalWrites": [],  # 未確認effectを隠す
})["ok"]
assert not verify_compaction(before, {
    **after,
    "completionChecks": [],  # 完了条件を落とす
})["ok"]
assert not verify_compaction(before, {
    **after,
    "nextExactAction": "公開する && 通知する",  # exact oneではない
})["ok"]

成功条件は、危険な欠落がすべて実装前に拒否され、元contextとcheckpointが保持されることです。dry-runではmanifestのfield名と差分理由だけを出し、会話本文やsecretをログへ出しません。

テストではhash一致だけで終わらず、再開後の最初の操作がnextExactActionと一致すること、完了済みphaseを二重実行しないこと、外部write countが増えないことまで確認します。

まとめ

文脈圧縮の安全性は、自然な要約文ではなく、再開に必要な決定と制約が残ったかで判断します。正本、完了phase、次の一手、外部effect、消費済み承認、禁止事項、rollback、完了条件をmanifest化し、semantic checksumと不変条件で照合します。

候補要約を検査してからatomicに昇格し、欠落時は元checkpointへ戻る。これによりAIエージェントは、長い履歴を小さくしても「忘れてはいけないもの」まで捨てず、安全に次のrunへ進めます。