成功し続ける自動化が、いつの間にか価値を失うとき — 成果の痩せを検出して畳む設計

import { appendFile } from "node:fs/promises";
 
type Outcome = {
  taskId: string;
  ranAt: string;      // ISO 8601 (UTC)
  status: "ok" | "fail";
  valueMetric: number; // その回の一次的な成果量
  evidence: string;    // 価値の根拠（出力先や要約）
  notes?: string;
};
 
export async function recordOutcome(o: Omit<Outcome, "ranAt">): Promise<void> {
  const line = JSON.stringify({ ...o, ranAt: new Date().toISOString() });
  // 追記専用。評価器はこの台帳だけを読む。
  await appendFile(process.env.OUTCOME_LEDGER ?? "./ledger.jsonl", line + "\n");
}
 
// 呼び出し例（スクリーンショット差分を検出するタスクの末尾）
const changed = await diffScreenshots();
await recordOutcome({
  taskId: "visual-regression",
  status: "ok",
  valueMetric: changed.length,          // 検出した差分の数がこのタスクの価値
  evidence: changed.length ? changed.slice(0, 3).join(", ") : "no diff",
});

{"taskId":"asset-audit","ranAt":"2026-06-30T01:00:00Z","status":"ok","valueMetric":11,"evidence":"11 assets flagged"}
{"taskId":"asset-audit","ranAt":"2026-07-01T01:00:00Z","status":"ok","valueMetric":4,"evidence":"4 assets flagged"}
{"taskId":"asset-audit","ranAt":"2026-07-02T01:00:00Z","status":"ok","valueMetric":1,"evidence":"1 asset flagged"}
{"taskId":"asset-audit","ranAt":"2026-07-03T01:00:00Z","status":"ok","valueMetric":0,"evidence":"nothing flagged"}
{"taskId":"asset-audit","ranAt":"2026-07-04T01:00:00Z","status":"ok","valueMetric":0,"evidence":"nothing flagged"}
{"taskId":"asset-audit","ranAt":"2026-07-05T01:00:00Z","status":"ok","valueMetric":0,"evidence":"nothing flagged"}

import json, sys
from collections import defaultdict
 
WINDOW = 9          # 直近何回を見るか
ZERO_STREAK = 5     # 連続ゼロがこの回数で退役候補
DECAY_RATIO = 0.4   # 後半平均 / 前半平均 がこの値未満なら「痩せ」
 
def load(path):
    runs = defaultdict(list)
    with open(path) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            r = json.loads(line)
            runs[r["taskId"]].append(r)
    return runs
 
def classify(recs):
    recent = recs[-WINDOW:]
    values = [r["valueMetric"] for r in recent]
    ok_rate = sum(1 for r in recent if r["status"] == "ok") / len(recent)
 
    trailing_zero = 0
    for v in reversed(values):
        if v == 0:
            trailing_zero += 1
        else:
            break
 
    half = max(1, len(values) // 2)
    first = sum(values[:half]) / half
    last = sum(values[half:]) / max(1, len(values) - half)
    decay = (last / first) if first > 0 else (0.0 if last == 0 else 1.0)
 
    if ok_rate < 1.0 and trailing_zero < ZERO_STREAK:
        return "repair"          # 落ちている→修理
    if trailing_zero >= ZERO_STREAK:
        return "retire?"         # 緑のまま連続ゼロ→退役を検討
    if first > 0 and decay < DECAY_RATIO:
        return "decaying"        # なだらかに痩せている→観察強化
    return "healthy"
 
def main(path):
    for task_id, recs in load(path).items():
        if len(recs) < 3:
            continue
        verdict = classify(recs)
        if verdict != "healthy":
            tail = [r["valueMetric"] for r in recs[-WINDOW:]]
            print(f"[{verdict:9}] {task_id:22} values={tail}")
 
if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "ledger.jsonl")

#!/usr/bin/env bash
# weekly-decay-review.sh — 週次で成果台帳を評価し、候補が出たら通知
set -euo pipefail
 
LEDGER="${OUTCOME_LEDGER:-$HOME/ops/ledger.jsonl}"
REPORT="$(python3 evaluate_decay.py "$LEDGER")"
 
if [ -n "$REPORT" ]; then
  # 退役候補・痩せ候補があるときだけ本文を送る（失敗のみ通知の思想と同じ）
  printf '成果が痩せた自動化の候補:\n%s\n' "$REPORT" | notify-ops --title "weekly decay review"
else
  echo "no decay candidates"   # 静かな週は静かに終える
fi