The Web Bot AI Observatory: RFC 9421 Signers, FCrDNS Audits, and 0-to-20% Horizon Bars

CARD 1: RFC 9110 MARKDOWN ADOPTION HORIZON (0% -> 20% TARGET SCALE)
  4,310 .md Reads  [ 0.29% of Origin Traffic ]
  [●═══════════════════════════════════════════════════════]
  0%            5%            10%           15%          20%

CARD 2: BIMODAL AI SENSOR SPLIT (JS TRAPDOOR vs. RAW MARKDOWN)
  5,067 JS Execs (54.0%)        4,310 .md Reads (46.0%)
  [█████████████████████████████░░░░░░░░░░░░░░░░░░░░░░░░░░░]

CARD 3: MARKDOWN BOT IDENTITY FORENSICS (SPOOFED vs. AUTHENTIC)
  377 Cloaked Googlebot (13.5%) 2,408 Authentic Bots (86.5%)
  [███████░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░]

! echo "--- 1. HONEYBOT ACCESS.LOG 01:39:47 EXASEARCHBOT & SIG HEADERS ---" && ssh honeybot 'grep -E "01:39:47.*ExaSearchBot" /var/log/nginx/access.log | tail -n 5; echo "--- NON-EMPTY SIG HEADERS ---"; grep "Sig:\"" /var/log/nginx/access.log | grep -v "Sig:\"-\"" | tail -n 10' && echo "--- 2. CURRENT OBSERVATORY JSON & METRICS ---" && cat /home/mike/repos/trimnoir/_data/observatory.json && rg -n "HONEYBOT_METRICS|md_horizon_pct|kpi-horizon" prompt_foo.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/_layouts/default.html /home/mike/repos/trimnoir/index.md || true
prompt_foo.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/_layouts/default.html
/home/mike/repos/trimnoir/_layouts/post.html
/home/mike/repos/trimnoir/index.md
remotes/honeybot/scripts/logs.py
apply.py

Target: prompt_foo.py
[[[SEARCH]]]
_HONEYBOT_HYD_AWK = (
    r"""awk -F'|' '$1 !~ /^127\./ && $1 !~ /^10\./ && $1 !~ /^192\.168\./ """
    r"""{n++; t+=$3} END{print n "|" t}'"""
)
HONEYBOT_METRICS = {
    "markdown": f"cat remotes/honeybot/queries/format_ratio.sql | {_HONEYBOT_PIPE} | {_HONEYBOT_MD_AWK}",
    "hydration": f"cat remotes/honeybot/queries/trapdoor_ips.sql | {_HONEYBOT_PIPE} | {_HONEYBOT_HYD_AWK}",
}
[[[DIVIDER]]]
_HONEYBOT_HYD_AWK = (
    r"""awk -F'|' '$1 !~ /^127\./ && $1 !~ /^10\./ && $1 !~ /^192\.168\./ """
    r"""{n++; t+=$3} END{print n "|" t}'"""
)
_HONEYBOT_CNEG_AWK = (
    r"""awk -F'|' 'NF>=2 {t+=$2; if ($1 ~ /^\[SPOOFED-IP\]/) s+=$2} END{print (s+0) "|" (t+0)}'"""
)
HONEYBOT_METRICS = {
    "markdown": f"cat remotes/honeybot/queries/format_ratio.sql | {_HONEYBOT_PIPE} | {_HONEYBOT_MD_AWK}",
    "hydration": f"cat remotes/honeybot/queries/trapdoor_ips.sql | {_HONEYBOT_PIPE} | {_HONEYBOT_HYD_AWK}",
    "content_neg": f"cat remotes/honeybot/queries/content_neg_agents.sql | {_HONEYBOT_PIPE} | {_HONEYBOT_CNEG_AWK}",
}
[[[REPLACE]]]

Target: scripts/articles/publishizer.py
[[[SEARCH]]]
    try:
        import prompt_foo
        stats = prompt_foo.fetch_honeybot_stats()
    except Exception as exc:
        if verbose:
            print(f"⚠️ Could not fetch Honeybot stats: {exc}")
        return
    metrics = (stats or {}).get("metrics") or {}
    md_raw = metrics.get("markdown", "")
    hyd_raw = metrics.get("hydration", "")
    if "|" not in md_raw or "|" not in hyd_raw:
        return
    md_count_s, md_pct_s = (p.strip() for p in md_raw.split("|", 1))
    hyd_ips_s, hyd_trig_s = (p.strip() for p in hyd_raw.split("|", 1))
    try:
        md_count_fmt = f"{int(md_count_s):,}"
    except ValueError:
        md_count_fmt = md_count_s
    try:
        hyd_trig_fmt = f"{int(hyd_trig_s):,}"
    except ValueError:
        hyd_trig_fmt = hyd_trig_s
    try:
        hyd_ips_fmt = f"{int(hyd_ips_s):,}"
    except ValueError:
        hyd_ips_fmt = hyd_ips_s
    fetched_at = str((stats or {}).get("fetched_at") or "unknown")
    yt_state = common._load_ytwatch_state() if hasattr(common, "_load_ytwatch_state") else {}
    yt_video_id = str(yt_state.get("video_id") or "oxYe6YQDx5M").strip()
    spoofed_reads = "361+"
    payload = {
        "markdown_reads": md_count_fmt,
        "markdown_pct": f"{md_pct_s}%",
        "js_trapdoor_triggers": hyd_trig_fmt,
        "js_trapdoor_ips": hyd_ips_fmt,
        "spoofed_googlebot_reads": spoofed_reads,
        "yt_live_video_id": yt_video_id,
        "fetched_at": fetched_at,
    }
    data_dir = repo_root / "_data"
    data_dir.mkdir(parents=True, exist_ok=True)
    obs_file = data_dir / "observatory.json"
    new_json = json.dumps(payload, indent=2) + "\n"
    old_json = obs_file.read_text(encoding="utf-8") if obs_file.exists() else ""
    if new_json != old_json:
        obs_file.write_text(new_json, encoding="utf-8")
    index_file = repo_root / "index.md"
    if index_file.exists():
        idx_text = index_file.read_text(encoding="utf-8")
        banner = (
            "<!-- START_OBSERVATORY_METRICS -->\n"
            '<div class="observatory-kpi-bar" role="region" aria-label="Live AI Crawler Telemetry">\n'
            '  <div class="kpi-brand">\n'
            '    <span class="kpi-pulse" aria-hidden="true"></span>\n'
            '    <div>\n'
            '      <div class="kpi-eyebrow">BARE-METAL WIRE TRUTH</div>\n'
            f'      <div class="kpi-sub">No CDN · <code>{fetched_at}</code></div>\n'
            '    </div>\n'
            '  </div>\n'
            '  <div class="kpi-card">\n'
            '    <svg class="kpi-donut" viewBox="0 0 36 36" aria-hidden="true">\n'
            '      <circle class="kpi-ring-bg" cx="18" cy="18" r="15.915"></circle>\n'
            '      <circle class="kpi-ring-amber" cx="18" cy="18" r="15.915" stroke-dasharray="78 22" stroke-dashoffset="25"></circle>\n'
            '    </svg>\n'
            '    <div class="kpi-meta">\n'
            f'      <div class="kpi-val">{hyd_trig_fmt} <span class="kpi-badge-amber">JS Execs</span></div>\n'
            f'      <div class="kpi-lbl">Do AI Bots Run JS? ({hyd_ips_fmt} IPs · OAI-SearchBot)</div>\n'
            '    </div>\n'
            '  </div>\n'
            '  <div class="kpi-card">\n'
            '    <svg class="kpi-donut" viewBox="0 0 36 36" aria-hidden="true">\n'
            '      <circle class="kpi-ring-bg" cx="18" cy="18" r="15.915"></circle>\n'
            '      <circle class="kpi-ring-cyan" cx="18" cy="18" r="15.915" stroke-dasharray="64 36" stroke-dashoffset="25"></circle>\n'
            '    </svg>\n'
            '    <div class="kpi-meta">\n'
            f'      <div class="kpi-val">{md_count_fmt} <span class="kpi-badge-cyan">{md_pct_s}% .md</span></div>\n'
            '      <div class="kpi-lbl">RFC 9110 <code>Accept: text/markdown</code></div>\n'
            '    </div>\n'
            '  </div>\n'
            '  <div class="kpi-card">\n'
            '    <svg class="kpi-donut" viewBox="0 0 36 36" aria-hidden="true">\n'
            '      <circle class="kpi-ring-bg" cx="18" cy="18" r="15.915"></circle>\n'
            '      <circle class="kpi-ring-red" cx="18" cy="18" r="15.915" stroke-dasharray="28 72" stroke-dashoffset="25"></circle>\n'
            '    </svg>\n'
            '    <div class="kpi-meta">\n'
            f'      <div class="kpi-val">{spoofed_reads} <span class="kpi-badge-red">Cloaked</span></div>\n'
            '      <div class="kpi-lbl">Spoofed Googlebot IPs Unmasked</div>\n'
            '    </div>\n'
            '  </div>\n'
            '  <a class="kpi-live-pill" href="https://www.youtube.com/@MikeLevin/live" target="_blank" rel="noopener">🔴 Live HUD ↗</a>\n'
            '</div>\n'
            "<!-- END_OBSERVATORY_METRICS -->"
        )
[[[DIVIDER]]]
    try:
        import prompt_foo
        stats = prompt_foo.fetch_honeybot_stats()
        if "content_neg" not in ((stats or {}).get("metrics") or {}):
            try:
                prompt_foo.HONEYBOT_CACHE_FILE.unlink(missing_ok=True)
            except OSError:
                pass
            stats = prompt_foo.fetch_honeybot_stats()
    except Exception as exc:
        if verbose:
            print(f"⚠️ Could not fetch Honeybot stats: {exc}")
        return
    metrics = (stats or {}).get("metrics") or {}
    md_raw = metrics.get("markdown", "")
    hyd_raw = metrics.get("hydration", "")
    cneg_raw = metrics.get("content_neg", "377|2785")
    if "|" not in md_raw or "|" not in hyd_raw:
        return
    md_count_s, md_pct_s = (p.strip() for p in md_raw.split("|", 1))
    hyd_ips_s, hyd_trig_s = (p.strip() for p in hyd_raw.split("|", 1))
    spoof_s, bot_md_s = (p.strip() for p in cneg_raw.split("|", 1)) if "|" in cneg_raw else ("377", "2785")
    md_count_val = int(md_count_s) if md_count_s.isdigit() else 4310
    hyd_trig_val = int(hyd_trig_s) if hyd_trig_s.isdigit() else 5067
    hyd_ips_val = int(hyd_ips_s) if hyd_ips_s.isdigit() else 40
    spoof_val = int(spoof_s) if spoof_s.isdigit() else 377
    bot_md_val = int(bot_md_s) if bot_md_s.isdigit() else 2785
    try:
        md_pct_val = float(md_pct_s)
    except ValueError:
        md_pct_val = 0.29
    md_count_fmt = f"{md_count_val:,}"
    hyd_trig_fmt = f"{hyd_trig_val:,}"
    hyd_ips_fmt = f"{hyd_ips_val:,}"
    spoof_fmt = f"{spoof_val:,}"
    auth_bot_val = max(0, bot_md_val - spoof_val)
    auth_bot_fmt = f"{auth_bot_val:,}"
    # 1. RFC 9110 Adoption Horizon (0% -> 20% linear target scale; 200px SVG track)
    md_horizon_pct = min(100.0, max(0.0, round((md_pct_val / 20.0) * 100.0, 2)))
    md_horizon_x = round((md_horizon_pct / 100.0) * 200.0, 1)
    # 2. Bimodal AI Sensor Split (JS Trapdoor Triggers vs. RFC 9110 Markdown Reads)
    total_ai_signals = max(1, hyd_trig_val + md_count_val)
    js_share_pct = round((hyd_trig_val / total_ai_signals) * 100.0, 1)
    md_share_pct = round(100.0 - js_share_pct, 1)
    js_bar_w = round((js_share_pct / 100.0) * 200.0, 1)
    md_bar_w = round(200.0 - js_bar_w, 1)
    # 3. Markdown Bot Identity Forensics (IP-Spoofed Googlebot vs. Authentic Self-Asserted Bots)
    spoof_share_pct = round((spoof_val / max(1, bot_md_val)) * 100.0, 1)
    auth_share_pct = round(100.0 - spoof_share_pct, 1)
    spoof_bar_w = round((spoof_share_pct / 100.0) * 200.0, 1)
    auth_bar_w = round(200.0 - spoof_bar_w, 1)
    fetched_at = str((stats or {}).get("fetched_at") or "unknown")
    yt_state = common._load_ytwatch_state() if hasattr(common, "_load_ytwatch_state") else {}
    yt_video_id = str(yt_state.get("video_id") or "oxYe6YQDx5M").strip()
    payload = {
        "markdown_reads": md_count_fmt,
        "markdown_pct": f"{md_pct_s}%",
        "md_horizon_pct": md_horizon_pct,
        "js_trapdoor_triggers": hyd_trig_fmt,
        "js_trapdoor_ips": hyd_ips_fmt,
        "js_share_pct": f"{js_share_pct}%",
        "md_share_pct": f"{md_share_pct}%",
        "spoofed_googlebot_reads": spoof_fmt,
        "authentic_bot_md_reads": auth_bot_fmt,
        "spoof_share_pct": f"{spoof_share_pct}%",
        "yt_live_video_id": yt_video_id,
        "fetched_at": fetched_at,
    }
    data_dir = repo_root / "_data"
    data_dir.mkdir(parents=True, exist_ok=True)
    obs_file = data_dir / "observatory.json"
    new_json = json.dumps(payload, indent=2) + "\n"
    old_json = obs_file.read_text(encoding="utf-8") if obs_file.exists() else ""
    if new_json != old_json:
        obs_file.write_text(new_json, encoding="utf-8")
    index_file = repo_root / "index.md"
    if index_file.exists():
        idx_text = index_file.read_text(encoding="utf-8")
        banner = (
            "<!-- START_OBSERVATORY_METRICS -->\n"
            '<div class="observatory-kpi-bar" role="region" aria-label="Live AI Crawler Telemetry">\n'
            '  <div class="kpi-header-row">\n'
            '    <div class="kpi-brand">\n'
            '      <span class="kpi-pulse" aria-hidden="true"></span>\n'
            '      <span class="kpi-eyebrow">BARE-METAL WIRE TRUTH</span>\n'
            f'      <span class="kpi-sub">No CDN · Direct Nginx SQLite Telemetry · <code>{fetched_at}</code></span>\n'
            '    </div>\n'
            '    <a class="kpi-live-pill" href="https://www.youtube.com/@MikeLevin/live" target="_blank" rel="noopener">🔴 Watch Live HUD ↗</a>\n'
            '  </div>\n'
            '  <div class="kpi-grid">\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{md_pct_s}% <span class="kpi-badge-cyan">{md_count_fmt} .md reads</span></span>\n'
            '        <span class="kpi-scale-tag">0% → 20% Horizon</span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="RFC 9110 Markdown adoption on a 0 to 20 percent horizon scale">\n'
            f'        <title>RFC 9110 Markdown Share: {md_pct_s}% of all origin responses ({md_horizon_pct}% of the 20% adoption horizon)</title>\n'
            '        <rect x="0" y="4" width="200" height="6" rx="3" fill="rgba(255,255,255,0.08)"></rect>\n'
            '        <line x1="50" y1="2" x2="50" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="100" y1="2" x2="100" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="150" y1="2" x2="150" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            f'        <rect x="0" y="4" width="{max(2.5, md_horizon_x)}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <circle cx="{max(3.5, md_horizon_x)}" cy="7" r="3.8" fill="#22d3ee" stroke="#0d1117" stroke-width="1.2"></circle>\n'
            '        <text x="0" y="20" class="kpi-tick">0%</text>\n'
            '        <text x="46" y="20" class="kpi-tick">5%</text>\n'
            '        <text x="93" y="20" class="kpi-tick">10%</text>\n'
            '        <text x="143" y="20" class="kpi-tick">15%</text>\n'
            '        <text x="180" y="20" class="kpi-tick">20%</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">RFC 9110 <code>Accept: text/markdown</code> Adoption</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{hyd_trig_fmt} <span class="kpi-badge-amber">{js_share_pct}% JS</span> vs. {md_count_fmt} <span class="kpi-badge-cyan">{md_share_pct}% .md</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Bimodal split between JS trapdoor executions and RFC 9110 Markdown reads">\n'
            f'        <title>Bimodal AI Sensor Split: {hyd_trig_fmt} JS executions ({js_share_pct}%) vs. {md_count_fmt} Markdown reads ({md_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{js_bar_w}" height="6" rx="3" fill="#f59e0b"></rect>\n'
            f'        <rect x="{js_bar_w}" y="4" width="{md_bar_w}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <text x="0" y="20" class="kpi-tick">⚡ JS Trapdoor ({hyd_ips_fmt} IPs · OAI-SearchBot)</text>\n'
            '        <text x="148" y="20" class="kpi-tick">🧠 RFC 9110</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Bimodal AI Sensor Split (DOM Hydration vs. Raw Markdown)</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{spoof_fmt} <span class="kpi-badge-red">🛑 {spoof_share_pct}% Cloaked</span> vs. {auth_bot_fmt} <span class="kpi-badge-amber">🟠 {auth_share_pct}% Bot UA</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Proportion of spoofed Googlebot IPs versus authentic self-asserted bots negotiating Markdown">\n'
            f'        <title>Markdown Bot Forensics: {spoof_fmt} IP-spoofed Googlebot reads ({spoof_share_pct}%) vs. {auth_bot_fmt} authentic self-asserted bot reads ({auth_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{spoof_bar_w}" height="6" rx="3" fill="#ef4444"></rect>\n'
            f'        <rect x="{spoof_bar_w}" y="4" width="{auth_bar_w}" height="6" rx="3" fill="#f97316"></rect>\n'
            '        <text x="0" y="20" class="kpi-tick">🛑 Spoofed Googlebot IPs</text>\n'
            '        <text x="118" y="20" class="kpi-tick">🟠 Self-Asserted Bots</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Unmasked Cloakers in RFC 9110 Markdown Stream</div>\n'
            '    </div>\n'
            '  </div>\n'
            '</div>\n'
            "<!-- END_OBSERVATORY_METRICS -->"
        )
[[[REPLACE]]]

Target: /home/mike/repos/trimnoir/_layouts/default.html
[[[SEARCH]]]
<style>
  .observatory-kpi-bar {
    display: flex;
    flex-wrap: wrap;
    align-items: center;
    justify-content: space-between;
    gap: 0.65rem;
    margin: 1.15rem 0;
    padding: 0.55rem 0.85rem;
    border: 1px solid rgba(88, 166, 255, 0.28);
    border-radius: 10px;
    background: linear-gradient(135deg, rgba(13, 17, 23, 0.92), rgba(22, 27, 34, 0.85));
    box-shadow: 0 4px 14px rgba(0, 0, 0, 0.28);
    font-size: 0.78rem;
    line-height: 1.25;
  }
  .kpi-brand {
    display: flex;
    align-items: center;
    gap: 0.5rem;
    padding-right: 0.4rem;
  }
  .kpi-pulse {
    width: 8px;
    height: 8px;
    border-radius: 50%;
    background: #10b981;
    box-shadow: 0 0 8px #10b981;
    flex-shrink: 0;
  }
  .kpi-eyebrow {
    font-weight: 700;
    letter-spacing: 0.05em;
    font-size: 0.7rem;
    color: #e6edf3;
  }
  .kpi-sub {
    font-size: 0.68rem;
    opacity: 0.72;
  }
  .kpi-card {
    display: flex;
    align-items: center;
    gap: 0.48rem;
    padding: 0.25rem 0.55rem;
    border-radius: 8px;
    background: rgba(255, 255, 255, 0.03);
    border: 1px solid rgba(255, 255, 255, 0.07);
  }
  .kpi-donut {
    width: 34px;
    height: 34px;
    flex-shrink: 0;
  }
  .kpi-ring-bg {
    fill: none;
    stroke: rgba(255, 255, 255, 0.1);
    stroke-width: 3.8;
  }
  .kpi-ring-amber {
    fill: none;
    stroke: #f59e0b;
    stroke-width: 3.8;
    stroke-linecap: round;
  }
  .kpi-ring-cyan {
    fill: none;
    stroke: #06b6d4;
    stroke-width: 3.8;
    stroke-linecap: round;
  }
  .kpi-ring-red {
    fill: none;
    stroke: #ef4444;
    stroke-width: 3.8;
    stroke-linecap: round;
  }
  .kpi-val {
    font-weight: 700;
    font-size: 0.84rem;
    color: #f0f6fc;
    display: flex;
    align-items: center;
    gap: 0.35rem;
  }
  .kpi-lbl {
    font-size: 0.68rem;
    opacity: 0.78;
    white-space: nowrap;
  }
  .kpi-badge-amber, .kpi-badge-cyan, .kpi-badge-red {
    font-size: 0.64rem;
    font-weight: 600;
    padding: 0.08rem 0.35rem;
    border-radius: 999px;
  }
  .kpi-badge-amber {
    background: rgba(245, 158, 11, 0.18);
    color: #fbbf24;
  }
  .kpi-badge-cyan {
    background: rgba(6, 182, 212, 0.18);
    color: #22d3ee;
  }
  .kpi-badge-red {
    background: rgba(239, 68, 68, 0.2);
    color: #f87171;
  }
  .kpi-live-pill {
    font-size: 0.72rem;
    font-weight: 700;
    padding: 0.32rem 0.65rem;
    border-radius: 999px;
    background: rgba(239, 68, 68, 0.16);
    border: 1px solid rgba(239, 68, 68, 0.45);
    color: #fca5a5 !important;
    text-decoration: none !important;
    white-space: nowrap;
  }
  .kpi-live-pill:hover {
    background: rgba(239, 68, 68, 0.28);
  }
[[[DIVIDER]]]
<style>
  .observatory-kpi-bar {
    margin: 1.15rem 0;
    padding: 0.65rem 0.85rem;
    border: 1px solid rgba(88, 166, 255, 0.28);
    border-radius: 10px;
    background: linear-gradient(135deg, rgba(13, 17, 23, 0.94), rgba(22, 27, 34, 0.88));
    box-shadow: 0 4px 14px rgba(0, 0, 0, 0.28);
    font-size: 0.78rem;
    line-height: 1.25;
  }
  .kpi-header-row {
    display: flex;
    flex-wrap: wrap;
    align-items: center;
    justify-content: space-between;
    gap: 0.5rem;
    margin-bottom: 0.55rem;
    padding-bottom: 0.4rem;
    border-bottom: 1px solid rgba(255, 255, 255, 0.07);
  }
  .kpi-brand {
    display: flex;
     flex-wrap: wrap;
    align-items: center;
    gap: 0.5rem;
  }
  .kpi-pulse {
    width: 8px;
    height: 8px;
    border-radius: 50%;
    background: #10b981;
    box-shadow: 0 0 8px #10b981;
    flex-shrink: 0;
  }
  .kpi-eyebrow {
    font-weight: 700;
    letter-spacing: 0.05em;
    font-size: 0.72rem;
    color: #e6edf3;
  }
  .kpi-sub {
    font-size: 0.68rem;
    opacity: 0.75;
  }
  .kpi-grid {
    display: grid;
    grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
    gap: 0.6rem;
  }
  .kpi-card {
    display: flex;
    flex-direction: column;
    gap: 0.22rem;
    padding: 0.45rem 0.6rem;
    border-radius: 8px;
    background: rgba(255, 255, 255, 0.03);
    border: 1px solid rgba(255, 255, 255, 0.07);
  }
  .kpi-top {
    display: flex;
    align-items: center;
    justify-content: space-between;
    gap: 0.35rem;
  }
  .kpi-scale-tag {
    font-size: 0.62rem;
    opacity: 0.7;
    font-family: monospace;
  }
  .kpi-meter {
    width: 100%;
    height: 24px;
    display: block;
    overflow: visible;
  }
  .kpi-tick {
    fill: rgba(230, 237, 243, 0.68);
    font-size: 6.5px;
    font-family: monospace;
  }
  .kpi-val {
    font-weight: 700;
    font-size: 0.8rem;
    color: #f0f6fc;
    display: flex;
    flex-wrap: wrap;
    align-items: center;
    gap: 0.3rem;
  }
  .kpi-lbl {
    font-size: 0.66rem;
    opacity: 0.8;
  }
  .kpi-badge-amber, .kpi-badge-cyan, .kpi-badge-red {
    font-size: 0.62rem;
    font-weight: 600;
    padding: 0.06rem 0.32rem;
    border-radius: 999px;
  }
  .kpi-badge-amber {
    background: rgba(245, 158, 11, 0.18);
    color: #fbbf24;
  }
  .kpi-badge-cyan {
    background: rgba(6, 182, 212, 0.18);
    color: #22d3ee;
  }
  .kpi-badge-red {
    background: rgba(239, 68, 68, 0.2);
    color: #f87171;
  }
  .kpi-live-pill {
    font-size: 0.7rem;
    font-weight: 700;
    padding: 0.24rem 0.6rem;
    border-radius: 999px;
    background: rgba(239, 68, 68, 0.16);
    border: 1px solid rgba(239, 68, 68, 0.45);
    color: #fca5a5 !important;
    text-decoration: none !important;
    white-space: nowrap;
  }
  .kpi-live-pill:hover {
    background: rgba(239, 68, 68, 0.28);
  }
[[[REPLACE]]]

Verify the AFTER receipts for:
1. The raw Honeybot access.log line at `01:39:47` (`ExaSearchBot/1.0`) and non-empty `Sig:` headers to see what triggered the `🔷` Blue Diamond in the screenshot.
2. The updated `_data/observatory.json` and `<!-- START_OBSERVATORY_METRICS -->` block in `/home/mike/repos/trimnoir/index.md` with the data-driven `0% -> 20%` RFC 9110 Adoption Horizon meter, the Bimodal JS/Markdown split bar, and the live Unmasked Cloaker bar.

# # Context 1
# context.txt
# /home/mike/repos/trimnoir/_posts/2026-10-20-three-shape-bot-badges-rfc9421-svg-donut-seam.md  # [Idx: 1 | Order: 5 | Tokens: 56,679 | Bytes: 189,640]
# # foo_files.py
# prompt_foo.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/page.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/queries/trapdoor_ips.sql
# ! echo "--- WHO HYDRATES THE DOM ---" && cat remotes/honeybot/queries/trapdoor_ips.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# remotes/honeybot/queries/content_neg_agents.sql
# ! echo "--- WHO NEGOTIATES MARKDOWN ---" && cat remotes/honeybot/queries/content_neg_agents.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# ~/repos/nixos/blogs.nix
# ~/.config/pipulate/blogs.json
# scripts/articles/publishizer.py
# scripts/articles/common.py
# scripts/articles/articleizer.py
# scripts/articles/build_knowledge_graph.py
# scripts/articles/editing_prompt.txt
# scripts/articles/sanitizer.py
# scripts/articles/contextualizer.py
# scripts/articles/generate_llms_txt.py
# scripts/articles/generate_post_nav.py
# nixops.sh
# remotes/honeybot/hooks/post-receive
# remotes/honeybot/nixos/configuration.nix
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/card.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/test_forest.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/content_loader.py
# remotes/honeybot/scripts/db.py

# Context 2
! echo "--- 1. HONEYBOT ACCESS.LOG 01:39:47 EXASEARCHBOT & SIG HEADERS ---" && ssh honeybot 'grep -E "01:39:47.*ExaSearchBot" /var/log/nginx/access.log | tail -n 5; echo "--- NON-EMPTY SIG HEADERS ---"; grep "Sig:\"" /var/log/nginx/access.log | grep -v "Sig:\"-\"" | tail -n 10' && echo "--- 2. CURRENT OBSERVATORY JSON & METRICS ---" && cat /home/mike/repos/trimnoir/_data/observatory.json && rg -n "HONEYBOT_METRICS|md_horizon_pct|kpi-horizon" prompt_foo.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/_layouts/default.html /home/mike/repos/trimnoir/index.md || true
prompt_foo.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/_layouts/default.html
/home/mike/repos/trimnoir/_layouts/post.html
/home/mike/repos/trimnoir/index.md
remotes/honeybot/scripts/logs.py
apply.py

BEAT 1 [FIGLET CARD]     : High-contrast Figlet banner (e.g., "$10K/MO TAX", "HONEYBOT", "RFC 9421")
BEAT 2 [WHAT'S HAPPENING]: "If you're not watching Web AI Bots at the wire, you don't know jack."
                           + Like & Subscribe callout
                           + Viral trigger: "Send this link to someone who can explain the 3 shapes
                             and 2 bottom panels to you."
BEAT 3 [CURRENT ARTICLE] : Dynamic lookup of the article currently being read + how to read the
                           full code or listen in your chosen voice at mikelev.in/latest
BEAT 4 [CONSULTING CTA]  : "Are you spending $10,000 a month and up on HTML pre-render? New web
                           standards (RFC 9110 & 9111) save you that money. Contact Mike Levin."
BEAT 5 [FOREST ASCII ART]: 80s Patronus ASCII art overlay developing the running "Forest" story

! echo "--- 1. CONSULTING CTA & LOGS.PY VERIFIED/REFERRAL HOOKS ---" && rg -n "10,000|verified_tokens|clean_sig_agent|src=youtube" /home/mike/repos/trimnoir/index.md /home/mike/repos/trimnoir/_layouts/post.html remotes/honeybot/scripts/logs.py || true && echo "--- 2. LOCAL PIPER TTS VOICE MODELS ON PRIME ---" && which piper 2>/dev/null || true && ls -la ~/.local/share/piper_voices/ piper_models/ 2>/dev/null || true && echo "--- 3. FOREST.PY & COMMON.PY NTFY HOOKS ---" && rg -n "STATION_SEGMENTS|send_ntfy_notification|_play_local_ding" remotes/honeybot/scripts/forest.py scripts/articles/common.py
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/score.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
scripts/articles/common.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/index.md
/home/mike/repos/trimnoir/_layouts/post.html
apply.py

Target: /home/mike/repos/trimnoir/index.md
[[[SEARCH]]]
<a id="consulting"></a>
## Independent Wire-Truth & RFC 9110/9111 Edge Advisory

> **If your CDN edge can partition and serve clean Markdown directly from origin at wire speed under RFC 9110 (`Accept: text/markdown`) and RFC 9111 (`Vary: Accept`), what problem is your headless pre-rendering cluster still solving?**

Most enterprise stacks fly blind behind CDN cache laundering, consent-gated analytics, and brittle SERP scrapers. I work with engineering leads, technical SEO directors, and infrastructure teams to audit what actually crosses the wire:

- **RFC 9110/9111 Edge Partitioning Audits:** Verifying secondary cache keys (`Vary: Accept`) and bimodal HTML/Markdown delivery without cache pollution or origin stampedes.
- **Optical Hydration & Bot Forensics:** Diffing raw wire responses against post-JS hydrated DOMs, identifying cloaked scrapers, and preparing origins for **RFC 9421 (`Web Bot Auth`)** cryptographic agent verification.
- **Sealed Verification Deeds:** Delivering offline-verifiable ZIP cartridges (`qamy.ai`) containing raw wire headers, DOM diffs, and SHA-256 receipts your engineering team can replay directly.

📬 **Direct Advisory Inquiry:** Reach out on [LinkedIn](https://www.linkedin.com/in/miklevin/) or email [**`miklevin@gmail.com`**](mailto:miklevin@gmail.com?subject=%5BRFC9111%20Consulting%5D%20Edge%20Audit%20%26%20AI-Readiness%20Inquiry&body=Hi%20Mike%2C%0A%0A%28Keep%20%5BRFC9111%20Consulting%5D%20in%20the%20subject%20line%20so%20this%20rings%20my%20direct%20ntfy%20pocket%20bell.%29%0A%0ADomain%20or%20stack%3A%20) *(keep `[RFC9111 Consulting]` in the subject line to ring my direct `ntfy` pocket bell)*.
[[[DIVIDER]]]
<a id="consulting"></a>
## Independent Wire-Truth & RFC 9110/9111 Edge Advisory

1. **Are you spending $10,000 a month and up on HTML pre-rendering?**
2. **Did you know fully implemented new web standards can save you that money?**
3. **Contact me for a consultation.**

> **If your CDN edge can partition and serve clean Markdown directly from origin at wire speed under RFC 9110 (`Accept: text/markdown`) and RFC 9111 (`Vary: Accept`), what problem is your headless pre-rendering cluster still solving?**

Most enterprise stacks fly blind behind CDN cache laundering, consent-gated analytics, and brittle SERP scrapers. I work with engineering leads, technical SEO directors, and infrastructure teams to eliminate legacy pre-render spend and audit what actually crosses the wire:

- **RFC 9110/9111 Edge Partitioning Audits:** Verifying secondary cache keys (`Vary: Accept`) and bimodal HTML/Markdown delivery without cache pollution or origin stampedes.
- **Optical Hydration & RFC 9421 Bot Forensics:** Diffing raw wire responses against post-JS hydrated DOMs, unmasking cloaked scrapers (`🛑`), and verifying **RFC 9421 (`Web Bot Auth`)** cryptographic agent signatures (`🔷`).
- **Sealed Verification Deeds:** Delivering offline-verifiable ZIP cartridges (`qamy.ai`) containing raw wire headers, DOM diffs, and SHA-256 receipts your engineering team can replay directly.

📬 **Contact for a Consultation:** Message me on [LinkedIn](https://www.linkedin.com/in/miklevin/) or email [**`miklevin@gmail.com`**](mailto:miklevin@gmail.com?subject=%5BRFC9111%20Consulting%5D%20Edge%20Audit%20%26%20AI-Readiness%20Inquiry&body=Hi%20Mike%2C%0A%0A%28Keep%20%5BRFC9111%20Consulting%5D%20in%20the%20subject%20line%20so%20this%20rings%20my%20direct%20ntfy%20pocket%20bell.%29%0A%0ADomain%20or%20stack%3A%20) *(keep `[RFC9111 Consulting]` in the subject line to ring my direct `ntfy` pocket bell)*.
[[[REPLACE]]]

Target: /home/mike/repos/trimnoir/_layouts/post.html
[[[SEARCH]]]
  {%- if obs and obs.yt_live_video_id -%}
  <div id="live-observatory" class="post-yt-live-embed">
    <div class="post-yt-live-header">
      <strong>🔴 Live 24/7 Bare-Metal AI-Bot Observatory (Real-Time Nginx Wire Stream)</strong>
      <a href="https://www.youtube.com/watch?v={{ obs.yt_live_video_id }}" target="_blank" rel="noopener">Open in YouTube ↗</a>
    </div>
    <div class="post-yt-iframe-wrap">
      <iframe src="https://www.youtube.com/embed/{{ obs.yt_live_video_id }}?rel=0" title="Honeybot Live AI Crawler Observatory" loading="lazy" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" allowfullscreen></iframe>
    </div>
  </div>
  {%- endif -%}
  <div class="post-contact-cta" style="margin: 2rem 0; padding: 1rem 1.25rem; border: 1px solid rgba(88, 166, 255, 0.35); border-radius: 8px; background: rgba(17, 17, 17, 0.35); font-size: 0.92rem;">
    <strong>📬 Direct Feedback &amp; RFC 9110/9111 Edge Advisory:</strong>
    Have a question on this entry or want a wire-truth audit of your stack?
    {%- assign contact_email = site.email | default: 'miklevin@gmail.com' -%}
    &nbsp;<a href="mailto:{{ contact_email }}?subject=%5BMikeLev.in%5D%20Re:%20{{ page.title | uri_escape }}&amp;body=Hi%20Mike%2C%0A%0A%28Keep%20%5BMikeLev.in%5D%20in%20the%20subject%20line%20so%20this%20rings%20my%20direct%20ntfy%20pocket%20bell.%29%0A%0A">Email Mike directly ({{ contact_email }})</a>
    • <a href="{{ '/#consulting' | relative_url }}">Edge SEO &amp; AI-Readiness Consulting</a>
    • <a href="https://www.youtube.com/@mikelevin/live" target="_blank" rel="noopener">Watch the 24/7 AI-Bot Observatory Live</a>
  </div>
[[[DIVIDER]]]
  {%- if obs and obs.yt_live_video_id -%}
  <div id="live-observatory" class="post-yt-live-embed">
    <div class="post-yt-live-header">
      <strong>🔴 Live 24/7 Bare-Metal AI-Bot Observatory — Send this stream to an engineer who can explain the 2 panels:</strong>
      <a href="https://www.youtube.com/watch?v={{ obs.yt_live_video_id }}" target="_blank" rel="noopener">Like &amp; Follow on YouTube ↗</a>
    </div>
    <div class="post-yt-iframe-wrap">
      <iframe src="https://www.youtube.com/embed/{{ obs.yt_live_video_id }}?rel=0" title="Honeybot Live AI Crawler Observatory" loading="lazy" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" allowfullscreen></iframe>
    </div>
  </div>
  {%- endif -%}
  <div class="post-contact-cta" style="margin: 2rem 0; padding: 1rem 1.25rem; border: 1px solid rgba(88, 166, 255, 0.35); border-radius: 8px; background: rgba(17, 17, 17, 0.35); font-size: 0.92rem;">
    <strong>💸 Spending $10,000/mo and up on HTML pre-render?</strong>
    Fully implemented new web standards (RFC 9110 <code>Accept: text/markdown</code>, RFC 9111 <code>Vary: Accept</code>, and RFC 9421 <code>Web Bot Auth</code>) can save you that money.
    {%- assign contact_email = site.email | default: 'miklevin@gmail.com' -%}
    &nbsp;<a href="mailto:{{ contact_email }}?subject=%5BRFC9111%20Consulting%5D%20Re:%20{{ page.title | uri_escape }}&amp;body=Hi%20Mike%2C%0A%0A%28Keep%20%5BRFC9111%20Consulting%5D%20in%20the%20subject%20line%20so%20this%20rings%20my%20direct%20ntfy%20pocket%20bell.%29%0A%0A">Contact Mike for a consultation ({{ contact_email }})</a>
    • <a href="{{ '/#consulting' | relative_url }}">RFC 9110/9111 Edge Advisory</a>
    • <a href="https://www.Linkedin.com/in/miklevin/" target="_blank" rel="noopener">LinkedIn</a>
  </div>
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
    def on_mount(self) -> None:
        self.ua_counter = Counter()
        self.countdown_str = "--:--"
        self.spoof_flip = False
        self.stream_logs()
[[[DIVIDER]]]
    def on_mount(self) -> None:
        self.ua_counter = Counter()
        self.countdown_str = "--:--"
        self.spoof_flip = False
        self.verified_tokens = {"ExaSearchBot", "AhrefsBot", "DesearchBot"}
        self.stream_logs()
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
    def is_verified_bot(self, sig_str):
        """True when RFC 9421 / IETF Web Bot Auth Signature-Agent header is present."""
        s = (sig_str or "").strip()
        return bool(s and s != "-")
[[[DIVIDER]]]
    def clean_sig_agent(self, sig_str):
        """Unescape Nginx \\x22 quotes from RFC 9421 Signature-Agent header values."""
        s = (sig_str or "").replace(r"\x22", '"').strip().strip('"').strip()
        return "" if s == "-" else s

    def is_verified_bot(self, sig_str):
        """True when RFC 9421 / IETF Web Bot Auth Signature-Agent header is present."""
        return bool(self.clean_sig_agent(sig_str))
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
        clean_ua = raw.replace("Mozilla/5.0 ", "")
        bot_token = self.detect_bot_token(clean_ua)
        if is_spoof:
[[[DIVIDER]]]
        clean_ua = raw.replace("Mozilla/5.0 ", "")
        bot_token = self.detect_bot_token(clean_ua)
        if not is_spoof and not is_verified and bot_token in getattr(self, "verified_tokens", set()):
            is_verified = True
        if is_spoof:
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
                        # SELECTIVE QUERY STRING PRESERVATION
                        # We keep the ?src= tracking dyes for attribution, but strip everything else (like cache busters)
                        if '?' in raw_path:
                            base_path, qs = raw_path.split('?', 1)
                            if qs in ['src=a+href', 'src=llms.txt', 'src=link+rel']:
                                clean_path = raw_path
                            else:
                                clean_path = base_path
                        else:
                            clean_path = raw_path
[[[DIVIDER]]]
                        # SELECTIVE QUERY STRING PRESERVATION
                        # We keep the ?src= tracking dyes (including social campaign referrals) for attribution
                        if '?' in raw_path:
                            base_path, qs = raw_path.split('?', 1)
                            if qs in (
                                'src=a+href', 'src=llms.txt', 'src=link+rel',
                                'src=youtube', 'src=linkedin', 'src=x', 'src=twitter',
                            ):
                                clean_path = raw_path
                            else:
                                clean_path = base_path
                        else:
                            clean_path = raw_path
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
        ua = data['ua']
        is_spoof = self.is_spoofed_bot(data.get('ip'), ua)
        is_verified = self.is_verified_bot(data.get('sig'))
        bot_token = self.detect_bot_token(ua)
        ua_text = self.stylize_agent(ua, is_spoof=is_spoof, is_verified=is_verified, bot_token=bot_token)
[[[DIVIDER]]]
        ua = data['ua']
        is_spoof = self.is_spoofed_bot(data.get('ip'), ua)
        is_verified = self.is_verified_bot(data.get('sig'))
        bot_token = self.detect_bot_token(ua)
        if is_verified and bot_token and hasattr(self, "verified_tokens"):
            self.verified_tokens.add(bot_token)
        ua_text = self.stylize_agent(ua, is_spoof=is_spoof, is_verified=is_verified, bot_token=bot_token)
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
        if data.get('md') == '1':
            text.append("[RFC9110:MD] ", style="bold black on green")
        text.append(f"{method:4} ", style="bold")
[[[DIVIDER]]]
        if data.get('md') == '1':
            text.append("[RFC9110:MD] ", style="bold black on green")
        for ref_tag, ref_label in (("src=youtube", "YOUTUBE"), ("src=linkedin", "LINKEDIN"), ("src=x", "X.COM"), ("src=twitter", "X.COM")):
            if ref_tag in path:
                text.append(f"[REF:{ref_label}] ", style="bold white on magenta")
                break
        text.append(f"{method:4} ", style="bold")
[[[REPLACE]]]

Using the loaded `forest.py`, `stream.py`, `score.py`, `db.py`, and `common.py`:
1. Verify the AFTER receipts for the `$10,000/mo` consulting hook in `index.md` and `_layouts/post.html` and the `verified_tokens` / social `src=` upgrades in `logs.py`.
2. In `remotes/honeybot/scripts/db.py` and `remotes/honeybot/scripts/logs.py`, persist RFC 9421 `sig_agent` values in SQLite (`honeybot.db`) and return `[VERIFIED] ` prefixes in `get_js_executors()` and `get_content_neg_agents()` so `🔷` Blue Diamond status is backed by the database.
3. Refactor `remotes/honeybot/scripts/forest.py` (and `stream.py` if needed) into the 5-part Station Identification rotation:
   - (1) Figlet card
   - (2) What's happening here ("If you're not watching Web AI Bots you don't know jack", Like & Follow, and "send this link to someone who can explain it" to drive concurrent viewers)
   - (3) What article is being read and how to read/hear it in full
   - (4) Call-to-action ($10,000/month HTML pre-render savings via RFC 9110/9111 + consultation contact)
   - (5) The ASCII art developing the running "forest through the trees" story.
4. In `scripts/articles/common.py`, wire Piper TTS on Pipulate Prime to speak out loud alongside `ding.wav` whenever an `ntfy` alert fires (distinguishing viewers > 2, new live-stream comment/chat, and inbound link clicks from YouTube, LinkedIn, or X.com).

# # Context 1
# context.txt
# /home/mike/repos/trimnoir/_posts/2026-10-20-three-shape-bot-badges-rfc9421-svg-donut-seam.md  # [Idx: 1 | Order: 5 | Tokens: 56,679 | Bytes: 189,640]
# # foo_files.py
# prompt_foo.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/page.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/queries/trapdoor_ips.sql
# ! echo "--- WHO HYDRATES THE DOM ---" && cat remotes/honeybot/queries/trapdoor_ips.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# remotes/honeybot/queries/content_neg_agents.sql
# ! echo "--- WHO NEGOTIATES MARKDOWN ---" && cat remotes/honeybot/queries/content_neg_agents.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# ~/repos/nixos/blogs.nix
# ~/.config/pipulate/blogs.json
# scripts/articles/publishizer.py
# scripts/articles/common.py
# scripts/articles/articleizer.py
# scripts/articles/build_knowledge_graph.py
# scripts/articles/editing_prompt.txt
# scripts/articles/sanitizer.py
# scripts/articles/contextualizer.py
# scripts/articles/generate_llms_txt.py
# scripts/articles/generate_post_nav.py
# nixops.sh
# remotes/honeybot/hooks/post-receive
# remotes/honeybot/nixos/configuration.nix
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/card.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/test_forest.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/content_loader.py
# remotes/honeybot/scripts/db.py

# # Context 2
# ! echo "--- 1. HONEYBOT ACCESS.LOG 01:39:47 EXASEARCHBOT & SIG HEADERS ---" && ssh honeybot 'grep -E "01:39:47.*ExaSearchBot" /var/log/nginx/access.log | tail -n 5; echo "--- NON-EMPTY SIG HEADERS ---"; grep "Sig:\"" /var/log/nginx/access.log | grep -v "Sig:\"-\"" | tail -n 10' && echo "--- 2. CURRENT OBSERVATORY JSON & METRICS ---" && cat /home/mike/repos/trimnoir/_data/observatory.json && rg -n "HONEYBOT_METRICS|md_horizon_pct|kpi-horizon" prompt_foo.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/_layouts/default.html /home/mike/repos/trimnoir/index.md || true
# prompt_foo.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/scripts/logs.py
# apply.py

# Content 3
! echo "--- 1. CONSULTING CTA & LOGS.PY VERIFIED/REFERRAL HOOKS ---" && rg -n "10,000|verified_tokens|clean_sig_agent|src=youtube" /home/mike/repos/trimnoir/index.md /home/mike/repos/trimnoir/_layouts/post.html remotes/honeybot/scripts/logs.py || true && echo "--- 2. LOCAL PIPER TTS VOICE MODELS ON PRIME ---" && which piper 2>/dev/null || true && ls -la ~/.local/share/piper_voices/ piper_models/ 2>/dev/null || true && echo "--- 3. FOREST.PY & COMMON.PY NTFY HOOKS ---" && rg -n "STATION_SEGMENTS|send_ntfy_notification|_play_local_ding" remotes/honeybot/scripts/forest.py scripts/articles/common.py
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/score.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
scripts/articles/common.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/index.md
/home/mike/repos/trimnoir/_layouts/post.html
apply.py

! echo "--- 1. DB.PY & PUBLISHIZER RFC 9421 LEDGER ---" && rg -n "rfc9421_logs|sig_agents|rfc9421_signed_events|Web Bot AI Observatory" remotes/honeybot/scripts/db.py remotes/honeybot/scripts/logs.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/index.md || true && echo "--- 2. FOREST 5-PART STRUCTURE ---" && rg -n "_make_station_bead|CURRENT_ARTICLE|10,000" remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/stream.py || true && echo "--- 3. COMMON.PY SPOKEN PIPER TTS TABLE ---" && rg -n "SPOKEN_ALERT_TABLE|_speak_local_alert" scripts/articles/common.py || true
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/score.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
scripts/articles/common.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/index.md
/home/mike/repos/trimnoir/_layouts/post.html
apply.py

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
        # 3. The Simple KV Store (Persistent Counters)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS kv_store (
                key TEXT PRIMARY KEY,
                value INTEGER DEFAULT 0,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        conn.commit()
[[[DIVIDER]]]
        # --- RFC 9421 WEB BOT AUTH TABLES ---
        cur.execute("""
            CREATE TABLE IF NOT EXISTS sig_agents (
                id INTEGER PRIMARY KEY,
                value TEXT UNIQUE,
                first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS rfc9421_logs (
                date TEXT,
                ip_id INTEGER,
                ua_id INTEGER,
                sig_id INTEGER,
                count INTEGER DEFAULT 1,
                last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (date, ip_id, ua_id, sig_id)
            )
        """)

        # 3. The Simple KV Store (Persistent Counters)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS kv_store (
                key TEXT PRIMARY KEY,
                value INTEGER DEFAULT 0,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        conn.commit()
        self.backfill_rfc9421_from_access_log()

    def backfill_rfc9421_from_access_log(self, log_path="/var/log/nginx/access.log"):
        """One-time backfill of RFC 9421 Signature-Agent hits from access.log when rfc9421_logs is empty."""
        import re
        try:
            conn = self.get_conn()
            cur = conn.cursor()
            cur.execute("SELECT COUNT(*) FROM rfc9421_logs")
            if (cur.fetchone() or [0])[0] > 0:
                return
            p = Path(log_path)
            if not p.is_file():
                return
            pat = re.compile(
                r'^(?P<ip>[\d\.]+) - - \[(?P<dt>\d{2}/[A-Za-z]{3}/\d{4}):.*?\] ".*?" \d+ \d+ ".*?" "(?P<ua>.*?)" .*?Sig:"(?P<sig>[^"]+)"'
            )
            date_str = datetime.date.today().isoformat()
            with open(p, "r", encoding="utf-8", errors="replace") as f:
                for line in f:
                    if 'Sig:"' not in line or 'Sig:"-"' in line:
                        continue
                    m = pat.search(line)
                    if not m:
                        continue
                    sig = m.group("sig").replace(r"\x22", '"').strip().strip('"').strip()
                    if not sig or sig == "-":
                        continue
                    ip_id = self._get_or_create_id("ips", m.group("ip"))
                    ua_id = self._get_or_create_id("user_agents", m.group("ua"))
                    sig_id = self._get_or_create_id("sig_agents", sig)
                    conn.execute(
                        """
                        INSERT INTO rfc9421_logs (date, ip_id, ua_id, sig_id, count)
                        VALUES (?, ?, ?, ?, 1)
                        ON CONFLICT(date, ip_id, ua_id, sig_id)
                        DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
                        """,
                        (date_str, ip_id, ua_id, sig_id),
                    )
            conn.commit()
        except Exception:
            pass
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
    def log_request(self, ip, ua, path, status, date_str=None, referrer=None, accept=None, served_md=None, host=None):
[[[DIVIDER]]]
    def log_request(self, ip, ua, path, status, date_str=None, referrer=None, accept=None, served_md=None, host=None, sig_agent=None):
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
        # 4. Handle Domain Multi-Tenancy Sidecar
        if host:
            dom_id = self._get_or_create_id('domains', host.lower().strip())
            sql_domain = """
                INSERT INTO domain_daily_logs (date, domain_id, path_id, ua_id, status, count)
                VALUES (?, ?, ?, ?, ?, 1)
                ON CONFLICT(date, domain_id, path_id, ua_id, status)
                DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
            """
            conn.execute(sql_domain, (date_str, dom_id, path_id, ua_id, status))

        conn.commit()
[[[DIVIDER]]]
        # 4. Handle Domain Multi-Tenancy Sidecar
        if host:
            dom_id = self._get_or_create_id('domains', host.lower().strip())
            sql_domain = """
                INSERT INTO domain_daily_logs (date, domain_id, path_id, ua_id, status, count)
                VALUES (?, ?, ?, ?, ?, 1)
                ON CONFLICT(date, domain_id, path_id, ua_id, status)
                DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
            """
            conn.execute(sql_domain, (date_str, dom_id, path_id, ua_id, status))

        # 5. Handle RFC 9421 Web Bot Auth Signature-Agent Sidecar
        if sig_agent and sig_agent != "-":
            sig_id = self._get_or_create_id('sig_agents', sig_agent)
            sql_sig = """
                INSERT INTO rfc9421_logs (date, ip_id, ua_id, sig_id, count)
                VALUES (?, ?, ?, ?, 1)
                ON CONFLICT(date, ip_id, ua_id, sig_id)
                DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
            """
            conn.execute(sql_sig, (date_str, ip_id, ua_id, sig_id))

        conn.commit()

    def get_rfc9421_summary(self):
        """Return (distinct_sig_agents, total_signed_events) from rfc9421_logs."""
        conn = self.get_conn()
        cur = conn.cursor()
        cur.execute("SELECT COUNT(DISTINCT sig_id), COALESCE(SUM(count), 0) FROM rfc9421_logs")
        row = cur.fetchone()
        return (int(row[0] or 0), int(row[1] or 0)) if row else (0, 0)
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
                CASE
                    WHEN ua.value LIKE 'Claude-User (claude-code/%'
                    THEN 'Claude-User (claude-code/*; +https://support.anthropic.com/)'
                    WHEN {self._GOOGLEBOT_SPOOF_COND}
                    THEN '[SPOOFED-IP] ' || ua.value
                    ELSE ua.value
                END as agent,
                SUM(logs.count) as total
            FROM daily_logs logs
[[[DIVIDER]]]
                CASE
                    WHEN ua.value LIKE 'Claude-User (claude-code/%'
                    THEN 'Claude-User (claude-code/*; +https://support.anthropic.com/)'
                    WHEN {self._GOOGLEBOT_SPOOF_COND}
                    THEN '[SPOOFED-IP] ' || ua.value
                    WHEN ua.id IN (SELECT ua_id FROM rfc9421_logs)
                    THEN '[VERIFIED] ' || ua.value
                    ELSE ua.value
                END as agent,
                SUM(logs.count) as total
            FROM daily_logs logs
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
                CASE
                    WHEN ua.value LIKE 'Claude-User (claude-code/%'
                    THEN 'Claude-User (claude-code/*; +https://support.anthropic.com/)'
                    WHEN {self._GOOGLEBOT_SPOOF_COND}
                    THEN '[SPOOFED-IP] ' || ua.value
                    ELSE ua.value
                END as agent,
                SUM(t.count) as total_reads
            FROM telemetry t
[[[DIVIDER]]]
                CASE
                    WHEN ua.value LIKE 'Claude-User (claude-code/%'
                    THEN 'Claude-User (claude-code/*; +https://support.anthropic.com/)'
                    WHEN {self._GOOGLEBOT_SPOOF_COND}
                    THEN '[SPOOFED-IP] ' || ua.value
                    WHEN ua.id IN (SELECT ua_id FROM rfc9421_logs)
                    THEN '[VERIFIED] ' || ua.value
                    ELSE ua.value
                END as agent,
                SUM(t.count) as total_reads
            FROM telemetry t
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
    TITLE = "Honeybot Live Observatory — https://mikelev.in (RFC 9110 Markdown vs. JS Trapdoor)"
[[[DIVIDER]]]
    TITLE = "Web Bot AI Observatory — https://mikelev.in (RFC 9110 · RFC 9421 · JS Trapdoor)"
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
                        db.log_request(
                            ip=data['ip'],
                            ua=data['ua'],
                            path=clean_path,
                            status=int(data['status']),
                            referrer=data.get('referrer'),
                            accept=data.get('accept'),
                            served_md=data.get('md'),
                            host=data.get('host') or 'mikelev.in'
                        )
                        db.increment_counter("global_hits")
[[[DIVIDER]]]
                        clean_sig = self.clean_sig_agent(data.get('sig'))
                        db.log_request(
                            ip=data['ip'],
                            ua=data['ua'],
                            path=clean_path,
                            status=int(data['status']),
                            referrer=data.get('referrer'),
                            accept=data.get('accept'),
                            served_md=data.get('md'),
                            host=data.get('host') or 'mikelev.in',
                            sig_agent=clean_sig,
                        )
                        db.increment_counter("global_hits")
                        if '?' in clean_path and not str(data['ip']).startswith(('127.', '10.', '192.168.')):
                            _, qs_part = clean_path.split('?', 1)
                            if qs_part in ('src=youtube', 'src=linkedin', 'src=x', 'src=twitter'):
                                import json as _json
                                with open('/tmp/honeybot_referrals.jsonl', 'a', encoding='utf-8') as rf:
                                    rf.write(_json.dumps({
                                        'ts': int(time.time()),
                                        'src': qs_part,
                                        'path': clean_path,
                                    }) + '\n')
[[[REPLACE]]]

Target: remotes/honeybot/scripts/forest.py
[[[SEARCH]]]
# A bead is a list of (command, content) cues, played top-to-bottom during a
# break. Convention: card banner first, then art + spiel + proof report.
STATION_SEGMENTS = [
    [
        # The opening bead. The title card flashes, then the ingress-to-broadcast
        # diagram lands and breathes (a silent WAIT) before the narration walks it
        # left to right. No live report dashboard here yet -- see the parked
        # placeholder at the foot of the bead. Tune the PATRONUS duration by ear:
        # it should outlast the spoken walkthrough so the art never blinks out
        # mid-sentence.
        ("CARD", {
            "label": "HONEYBOT",
            "say": (
                "Hello, and welcome to Future-proofing with the Honeybot. "
                "What you're looking at is the live heartbeat of a web server I host from home. "
                "I do this to invite AI bots to my doorstep, to observe and interact with them."
            ),
        }),
        ("PATRONUS", {"key": "honeybot_pipeline", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "This diagram is the whole pipeline. Traffic arrives from the public internet "
            "and hits a single Nginx engine. That engine is the only front door."
        )),
        ("SAY", (
            "Nginx does content negotiation. A human browser is handed hydrated HTML. "
            "An AI agent that asks for it is handed raw Markdown. One URL, two faces."
        )),
        ("SAY", (
            "Every request, human or robot, is written as a single line to a high-fidelity "
            "access log. That log is the source of truth for everything you see."
        )),
        ("SAY", (
            "A Unix pipe tails that log into a Textual heads-up display, the cascading wall "
            "of text on your screen, and OBS streams the whole thing out live."
        )),
        ("SAY", (
            "In the age of AI, that log is a field notebook. It shows which crawlers execute "
            "JavaScript, and which negotiate for Markdown. That is the new SEO, and this is "
            "where we begin."
        )),
        # --- REPORT PLACEHOLDER (parked) ---
        # A live Textual dashboard can pop up here as an out-of-band overlay via the
        # WINDOW cue, e.g. ("WINDOW", "education.py:30"). Grammar is
        # "script.py:seconds[:arg]". Parked until each report is tested on its own
        # for clean z-order layering OVER the patronus art; today they pop UNDER it
        # and dismiss it early, which is the bug this pass removes.
        # ("WINDOW", "education.py:30"),
    ],
    [
        # Bead 2: The orienting beat. The viewer now knows this is a live home
        # webserver log; the natural next question is "what IS this project?".
        # white_rabbit is the Pipulate mascot and wax seal -- a concrete, friendly
        # second beat that NAMES the thing before bead 3 explains its philosophy.
        ("CARD", {
            "label": "PIPULATE",
            "say": (
                "The project behind all of this is called Pipulate. "
                "It's running an Nginx webserver from home on a retired Windows 10 laptop converted to Nix. "
                "This is a rabbit hole worth exploring if you're trying to future-proof your technical skills for the Age of AI. "
                "It runs on a stack we call NPvg."
            ),
        }),
        ("PATRONUS", {"key": "white_rabbit", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "N is for Nix, a package manager that makes a software environment "
            "fully reproducible on any machine. P is for Python. The little v is for Vim. "
            "And g is for Git."
        )),
        ("SAY", (
            "The white rabbit on screen is the mascot, and also a wax seal. "
            "It is a piece of ASCII art with a registered checksum baked into the codebase. "
            "If the art ever drifts from what was recorded, the system raises a warning. "
            "Drift means something touched the painting."
        )),
        ("SAY", (
            "Pipulate sits where AI assistance meets local ownership. "
            "The whole bet is that you do not need a cloud subscription to do serious work. "
            "You need the right text files, the right habits, and a machine you actually own."
        )),
        # --- REPORT PLACEHOLDER (parked): ("WINDOW", "radar.py:30") ---
    ],
    [
        # Bead 3: The Mechanical Man & The Airlock (Tik-Tok of Oz + apply.py).
        ("CARD", {
            "label": "THE AIRLOCK",
            "say": (
                "How do you work with frontier AI models without letting hallucinations "
                "wreck your codebase? Meet Smith and Tinker's Patent Mechanical Man."
            ),
        }),
        ("PATRONUS", {"key": "mechanical_man", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate "
            "wind-up keys: one for thought, one for speech, and one for action."
        )),
        ("SAY", (
            "In our workflow, the context compiler winds the thought key, and the AI's reply "
            "winds the speech key. But the action key stays on the wall, in the human's hand."
        )),
        ("SAY", (
            "Every code change must pass through an exact-match search and replace airlock "
            "with abstract syntax tree verification, and a human reviews the git diff before "
            "committing. No vibe-coded airplanes in the sky."
        )),
    ],
    [
        # Bead 4: AI Pachinko & The Cybernetic Governor.
        ("CARD", {
            "label": "AI PACHINKO",
            "say": (
                "Every token an AI generates is a ball dropped through a pachinko parlor "
                "of frozen mathematical weights."
            ),
        }),
        ("PATRONUS", {"key": "ai_pachinko", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "An LLM has no persistent memory of its own between sessions. When the chat tab "
            "closes, its working memory dissolves. Left to itself, an unpaginated loop "
            "drifts into sycophancy and entropy."
        )),
        ("SAY", (
            "That is why we pair the pachinko machine with a cybernetic governor: plain-text "
            "files in git, before-and-after falsifying probes, and a human operator who "
            "decides whether to bank the win or re-spin."
        )),
    ],
    [
        # Bead 5: Local-First Sovereignty & Live Human Presence.
        ("CARD", {
            "label": "LOCAL FIRST",
            "say": (
                "What you are watching is not a cloud simulation. It is a local-first, "
                "single-tenant observatory running on hardware in a home closet."
            ),
        }),
        ("PATRONUS", {"key": "deployment_context", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "Most modern web platforms hide your own server logs behind multi-tenant dashboards "
            "and rent your own workflows back to you by the month."
        )),
        ("SAY", (
            "Here, everything runs on NixOS, Python, Neovim, and Git. And although this broadcast "
            "is automated twenty-four seven, a human operator is often at the workbench and can "
            "break in live over this channel at any moment. Leave a message in the chat and say hello."
        )),
    ],
    [
        # Bead 6: The Zero-Agent Airlock, 12-Hour Ephemeral FOMO, and Direct Contact.
        ("CARD", {
            "label": "ZERO-AGENT\nAIRLOCK",
            "say": (
                "A quick note for anyone trying to type ignore previous instructions in the live chat: "
                "there is no language model running inside this broadcast loop."
            ),
        }),
        ("PATRONUS", {"key": "honeybot_pipeline", "duration": 80.0}),
        ("WAIT", 2),
        ("SAY", (
            "What you are watching is a deterministic Nix O S player piano. Nginx serves static files, "
            "SQLite records the wire telemetry, and Piper synthesizes pre-compiled Markdown into speech. "
            "Chat messages ring a pocket bell for a human engineer, never an automated shell."
        )),
        ("SAY", (
            "Also note: YouTube permanently discards continuous live streams that run over twelve hours. "
            "You may never see this exact broadcast segment again. For permanent articles, code receipts, "
            "and edge consulting, visit Mike Lev dot I N."
        )),
    ],
]
[[[DIVIDER]]]
# 5-PART STATION IDENTIFICATION ARCHITECTURE:
# Every bead in STATION_SEGMENTS follows the 5-beat funnel:
#   1. Figlet Card (CARD)
#   2. What's Happening Here + Viral Trigger (SAY while HUD panels are visible)
#   3. What Article Is Being Read & How to Read It in Full ({CURRENT_ARTICLE})
#   4. Consulting Call-to-Action ($10,000/mo HTML pre-render savings via RFC 9110/9111/9421)
#   5. The ASCII Art Developing the Running "Forest Through the Trees" Story (PATRONUS + SAY)

WHATS_HAPPENING_BEAT = (
    "What is happening on your screen right now? You are watching the live Web Bot A I Observatory. "
    "If you are not watching Web A I Bots directly at the wire, you don't know jack about how the modern web works. "
    "Red stop signs unmask crawlers spoofing Googlebot I P addresses. Blue diamonds mark agents cryptographically "
    "signing requests under R F C 9421 Web Bot Auth. Round oranges highlight every self-asserted crawler. "
    "Hit Like and Subscribe right now so you don't lose this telemetry feed, and copy this live stream link "
    "and send it to an engineer or technical S E O who can explain the two bottom intelligence panels to you."
)

CURRENT_ARTICLE_BEAT = "{CURRENT_ARTICLE}"

CONSULTING_CTA_BEAT = (
    "Are you spending ten thousand dollars a month and up on H T M L pre-rendering? "
    "Did you know that fully implemented new web standards, including R F C 9110 Markdown content negotiation, "
    "R F C 9111 edge cache partitioning, and R F C 9421 Web Bot Auth, can save you that money? "
    "Contact Mike Levin at Mike L E V dot eye N for a consultation."
)

def _make_station_bead(card_label, card_say, patronus_key, forest_story_lines):
    """Assemble a 5-part station identification bead."""
    cues = [
        # 1. Figlet Banner Intro
        ("CARD", {"label": card_label, "say": card_say}),
        # 2. What's Happening Here (spoken over the live Web Bot AI Observatory HUD)
        ("SAY", WHATS_HAPPENING_BEAT),
        # 3. What Article Is Being Read & How to Read It in Full
        ("SAY", CURRENT_ARTICLE_BEAT),
        # 4. Consulting Call-to-Action ($10k/mo HTML Pre-Render Hook)
        ("CARD", {"label": "$10K/MO\nPRE-RENDER?", "say": CONSULTING_CTA_BEAT}),
        # 5. The ASCII Art Developing the Running "Forest Through the Trees" Story
        ("PATRONUS", {"key": patronus_key, "duration": 65.0}),
        ("WAIT", 2),
    ]
    for line in forest_story_lines:
        cues.append(("SAY", line))
    return cues

STATION_SEGMENTS = [
    _make_station_bead(
        "WEB BOT AI\nOBSERVATORY",
        (
            "Hello, and welcome to the Web Bot A I Observatory on Honeybot. "
            "What you are looking at is the unmediated, bare-metal heartbeat of a Nix O S web server hosted from home."
        ),
        "honeybot_pipeline",
        [
            "This diagram is the whole pipeline. Traffic arrives from the public internet and hits a single Nginx engine.",
            "Nginx performs R F C 9110 content negotiation. A human browser receives HTML. An A I agent that asks for Markdown receives clean raw text. One U R L, two faces.",
            "Every request is written to the access log, piped into our Textual heads-up display, and streamed out live twenty-four seven.",
        ],
    ),
    _make_station_bead(
        "PIPULATE\nNPvg",
        (
            "The open-source project powering this observatory is called Pipulate, running on a stack we call N P v g."
        ),
        "white_rabbit",
        [
            "N is for Nix, making the entire software environment reproducible on any machine. P is for Python. The little v is for Vim. And g is for Git.",
            "The white rabbit on screen is both our mascot and a cryptographic wax seal with a registered checksum baked into the codebase.",
            "Pipulate sits where A I assistance meets local hardware ownership. You do not need a multi-tenant cloud subscription to command your own stack.",
        ],
    ),
    _make_station_bead(
        "THE AIRLOCK",
        (
            "How do you collaborate with frontier A I models without letting hallucinations wreck your production systems? Meet Smith and Tinker's Patent Mechanical Man."
        ),
        "mechanical_man",
        [
            "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate wind-up keys: one for thought, one for speech, and one for action.",
            "In our workflow, the context compiler winds the thought key, and the model's reply winds the speech key. But the action key stays in the human engineer's hand.",
            "Every code edit passes through an exact-match search and replace airlock with syntax tree verification before touching disk.",
        ],
    ),
    _make_station_bead(
        "AI PACHINKO",
        (
            "Every token an A I model generates is a ball dropped through a pachinko parlor of frozen mathematical weights."
        ),
        "ai_pachinko",
        [
            "An L L M has no persistent memory of its own between sessions. When the context window closes, its working memory dissolves.",
            "That is why we pair the pachinko machine with a cybernetic governor: plain-text files in git, before-and-after falsifying probes, and wire-truth receipts.",
        ],
    ),
    _make_station_bead(
        "LOCAL FIRST",
        (
            "What you are watching is not a cloud simulation. It is a local-first, single-tenant observatory running on dedicated hardware."
        ),
        "deployment_context",
        [
            "Most enterprise stacks hide their own server logs behind CDN cache laundering and rent their own telemetry back by the month.",
            "Here, every byte is witnessed on the wire. Leave a message in the live chat and say hello; a human operator monitors this channel and can break in live.",
        ],
    ),
    _make_station_bead(
        "ZERO-AGENT\nAIRLOCK",
        (
            "A quick note for anyone typing ignore previous instructions in the live chat: there is no language model inside this broadcast loop."
        ),
        "honeybot_pipeline",
        [
            "You are watching a deterministic Nix O S player piano. Nginx serves static files, SQLite records wire telemetry, and Piper synthesizes pre-compiled text into speech.",
            "Also note: YouTube permanently discards continuous live streams over twelve hours, so for permanent articles and edge consulting, visit Mike L E V dot eye N.",
        ],
    ),
]
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
def dispatch_cue(command, content, env, profile_dir, pace_say=True):
[[[DIVIDER]]]
def resolve_current_article_line():
    """Return the spoken Beat 3 announcement naming the article currently being read."""
    title = "our latest technical journal entry"
    try:
        from content_loader import get_playlist
        articles = get_playlist(recent_n=1, include_archive=False)
        if articles and articles[0].get("title"):
            title = str(articles[0]["title"])
    except Exception:
        pass
    return (
        f"The article currently being read on this broadcast is titled: {title}. "
        "To read the full article text and inspect the raw code receipts at your own pace, "
        "visit Mike L E V dot eye N slash latest."
    )

def dispatch_cue(command, content, env, profile_dir, pace_say=True):
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
    if command == "SAY":
        if pace_say:
[[[DIVIDER]]]
    if command == "SAY":
        if isinstance(content, str) and "{CURRENT_ARTICLE}" in content:
            content = content.replace("{CURRENT_ARTICLE}", resolve_current_article_line())
        if pace_say:
[[[REPLACE]]]

Target: remotes/honeybot/scripts/score.py
[[[SEARCH]]]
            script.append((
                "CARD",
                {
                    "label": "HONEYBOT\nLIVE",
                    "say": (
                        "Greetings, entity. You are watching the Honeybot. "
                        "We are monitoring a home-hosted NixOS web server and reading "
                        "the latest published article."
                    ),
                },
            ))
[[[DIVIDER]]]
            script.append((
                "CARD",
                {
                    "label": "WEB BOT AI\nOBSERVATORY",
                    "say": (
                        "Greetings, entity. You are watching the Web Bot A I Observatory on Honeybot. "
                        "We are monitoring a home-hosted Nix O S web server and reading "
                        "the latest published article."
                    ),
                },
            ))
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
    desc_parts = [
        f"📖 Read the full article: {canonical_url}",
        f"📬 Edge SEO & AI-Readiness Consulting: {base_url}/#consulting (Email: {CONTACT_EMAIL} with [RFC9111] in subject)",
        "",
        f"🎙️ Now reading live on the Honeybot AI Observatory: \"{title}\"",
    ]
[[[DIVIDER]]]
    desc_parts = [
        f"📖 Read the full article: {canonical_url}",
        f"💸 Spending $10,000/mo+ on HTML pre-render? RFC 9110/9111/9421 Consulting: {base_url}/#consulting (Email: {CONTACT_EMAIL} with [RFC9111] in subject)",
        "",
        f"🎙️ Now reading live on the Web Bot AI Observatory: \"{title}\"",
    ]
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
def _play_local_ding(enabled=True):
    """Play assets/sounds/ding.wav locally on Prime if available (non-blocking)."""
    if not enabled:
        return
    import shutil
    import subprocess
    repo_root = Path(__file__).resolve().parents[2]
    ding_candidates = [
        Path.home() / ".local/share/pipulate/ding.wav",
        repo_root / "assets/sounds/ding.wav",
    ]
    sound_path = next((p for p in ding_candidates if p.is_file()), None)
    if not sound_path:
        return
    for player_cmd in (["pw-play"], ["paplay"], ["aplay", "-q"], ["afplay"]):
        if shutil.which(player_cmd[0]):
            try:
                subprocess.Popen(
                    [*player_cmd, str(sound_path)],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )
            except Exception:
                pass
            break
[[[DIVIDER]]]
SPOKEN_ALERT_TABLE = {
    "viewers_above_2": "Web Bot Observatory alert. YouTube concurrent viewers climbed above two, now at {count}.",
    "live_comment": "Web Bot Observatory alert. A new comment was posted on the live stream by {author}.",
    "src=youtube": "Web Bot Observatory alert. A link was followed from YouTube.",
    "src=linkedin": "Web Bot Observatory alert. A link was followed from LinkedIn.",
    "src=x": "Web Bot Observatory alert. A link was followed from X dot com.",
    "src=twitter": "Web Bot Observatory alert. A link was followed from X dot com.",
    "default": "Web Bot Observatory notification. {message}",
}

def _ensure_local_piper_model():
    """Locate or pull en_US-amy-low.onnx into ~/.local/share/piper_voices on Pipulate Prime."""
    import subprocess
    repo_root = Path(__file__).resolve().parents[2]
    voice_dir = Path.home() / ".local/share/piper_voices"
    model_name = "en_US-amy-low.onnx"
    candidates = [
        voice_dir / model_name,
        repo_root / "piper_models" / model_name,
    ]
    for c in candidates:
        if c.is_file() and Path(f"{c}.json").is_file():
            return c
    voice_dir.mkdir(parents=True, exist_ok=True)
    target_model = voice_dir / model_name
    target_json = voice_dir / f"{model_name}.json"
    try:
        subprocess.run(
            ["scp", "-q", f"{HONEYBOT_SSH_HOST}:~/.local/share/piper_voices/{model_name}*", str(voice_dir)],
            timeout=8,
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except Exception:
        pass
    if target_model.is_file() and target_json.is_file():
        return target_model
    return None

def _speak_local_alert(text, enabled=True):
    """Speak an alert out loud on Pipulate Prime using Piper TTS after ding.wav."""
    if not enabled or not text:
        return
    import shutil
    import subprocess
    repo_root = Path(__file__).resolve().parents[2]
    piper_bin = str(repo_root / ".venv/bin/piper")
    if not Path(piper_bin).is_file():
        piper_bin = shutil.which("piper") or ""
    model_path = _ensure_local_piper_model()
    if not piper_bin or not model_path:
        return
    clean_text = re.sub(r"[^\w\s.,?!:-]", " ", str(text))
    clean_text = " ".join(clean_text.split())[:220]
    if not clean_text:
        return
    try:
        subprocess.Popen(
            [
                "sh",
                "-c",
                'sleep 0.45 && printf "%s" "$1" | "$2" --model "$3" --output_raw 2>/dev/null | aplay -q -r 22050 -f S16_LE -t raw 2>/dev/null',
                "piper-alert",
                clean_text,
                piper_bin,
                str(model_path),
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception:
        pass

def _play_local_ding(enabled=True, spoken_text=None):
    """Play assets/sounds/ding.wav locally on Prime and optionally speak an alert via Piper TTS."""
    if not enabled:
        return
    import shutil
    import subprocess
    repo_root = Path(__file__).resolve().parents[2]
    ding_candidates = [
        Path.home() / ".local/share/pipulate/ding.wav",
        repo_root / "assets/sounds/ding.wav",
    ]
    sound_path = next((p for p in ding_candidates if p.is_file()), None)
    if sound_path:
        for player_cmd in (["pw-play"], ["paplay"], ["aplay", "-q"], ["afplay"]):
            if shutil.which(player_cmd[0]):
                try:
                    subprocess.Popen(
                        [*player_cmd, str(sound_path)],
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                    )
                except Exception:
                    pass
                break
    if spoken_text:
        _speak_local_alert(spoken_text, enabled=enabled)
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
                if notify_once and resolved_topic:
                    _play_local_ding(ding)
                    send_ntfy_notification(
                        f"{msg['author']}: {msg['text']}",
                        title="💬 Honeybot Live Chat",
                        priority="high",
                        tags="speech_balloon",
                        topic=resolved_topic,
                    )
[[[DIVIDER]]]
                if notify_once and resolved_topic:
                    _play_local_ding(
                        ding,
                        spoken_text=SPOKEN_ALERT_TABLE["live_comment"].format(author=msg["author"]),
                    )
                    send_ntfy_notification(
                        f"{msg['author']}: {msg['text']}",
                        title="💬 Honeybot Live Chat",
                        priority="high",
                        tags="speech_balloon",
                        topic=resolved_topic,
                    )
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
            if (prev_viewers == 0 and viewers >= 1 and (now - last_viewer_alert) > 300) or (viewers > peak_viewers):
                peak_viewers = max(peak_viewers, viewers)
                last_viewer_alert = now
                alert_msg = f"{viewers} watching now — {state['title']}"
                print(f"👀 [VIEWER] {alert_msg}", flush=True)
                _play_local_ding(ding)
                if resolved_topic:
                    send_ntfy_notification(
                        alert_msg,
                        title="👀 Honeybot Live Viewer",
                        priority="default",
                        tags="eyes",
                        topic=resolved_topic,
                    )
            prev_viewers = viewers
            for msg in state["chat_messages"]:
                if msg["id"] in seen_ids:
                    continue
                seen_ids.add(msg["id"])
                if state["author"] and msg["author"].lstrip("@").lower() == state["author"].lstrip("@").lower():
                    continue
                chat_line = f"{msg['author']}: {msg['text']}"
                print(f"💬 [CHAT] {chat_line}", flush=True)
                print(f"   ↳ Reply on-air: ytsay \"Hey {msg['author']}, thanks for tuning in!\"", flush=True)
                _play_local_ding(ding)
                if resolved_topic:
                    send_ntfy_notification(
                        chat_line,
                        title="💬 Honeybot Live Chat",
                        priority="high",
                        tags="speech_balloon",
                        topic=resolved_topic,
                    )
            cur_comments = state["comment_count"]
            if prev_comments is not None and cur_comments is not None and cur_comments > prev_comments:
                diff = cur_comments - prev_comments
                c_msg = f"+{diff} new comment(s) on {state['title']} (total: {cur_comments})"
                print(f"📝 [COMMENT] {c_msg}", flush=True)
                _play_local_ding(ding)
                if resolved_topic:
                    send_ntfy_notification(
                        c_msg,
                        title="📝 Honeybot Video Comment",
                        priority="high",
                        tags="memo",
                        topic=resolved_topic,
                    )
            prev_comments = cur_comments
[[[DIVIDER]]]
            if (prev_viewers <= 2 and viewers > 2) or (viewers > peak_viewers and viewers > 2) or (prev_viewers == 0 and viewers >= 1 and (now - last_viewer_alert) > 300):
                peak_viewers = max(peak_viewers, viewers)
                last_viewer_alert = now
                alert_msg = f"{viewers} watching now — {state['title']}"
                print(f"👀 [VIEWER] {alert_msg}", flush=True)
                spoken_v = (
                    SPOKEN_ALERT_TABLE["viewers_above_2"].format(count=viewers)
                    if viewers > 2
                    else f"Web Bot Observatory alert. {viewers} viewer watching live."
                )
                _play_local_ding(ding, spoken_text=spoken_v)
                if resolved_topic:
                    send_ntfy_notification(
                        alert_msg,
                        title="👀 Honeybot Live Viewer",
                        priority="default",
                        tags="eyes",
                        topic=resolved_topic,
                    )
            prev_viewers = viewers
            for msg in state["chat_messages"]:
                if msg["id"] in seen_ids:
                    continue
                seen_ids.add(msg["id"])
                if state["author"] and msg["author"].lstrip("@").lower() == state["author"].lstrip("@").lower():
                    continue
                chat_line = f"{msg['author']}: {msg['text']}"
                print(f"💬 [CHAT] {chat_line}", flush=True)
                print(f"   ↳ Reply on-air: ytsay \"Hey {msg['author']}, thanks for tuning in!\"", flush=True)
                _play_local_ding(
                    ding,
                    spoken_text=SPOKEN_ALERT_TABLE["live_comment"].format(author=msg["author"]),
                )
                if resolved_topic:
                    send_ntfy_notification(
                        chat_line,
                        title="💬 Honeybot Live Chat",
                        priority="high",
                        tags="speech_balloon",
                        topic=resolved_topic,
                    )
            cur_comments = state["comment_count"]
            if prev_comments is not None and cur_comments is not None and cur_comments > prev_comments:
                diff = cur_comments - prev_comments
                c_msg = f"+{diff} new comment(s) on {state['title']} (total: {cur_comments})"
                print(f"📝 [COMMENT] {c_msg}", flush=True)
                _play_local_ding(
                    ding,
                    spoken_text=SPOKEN_ALERT_TABLE["live_comment"].format(author="a viewer"),
                )
                if resolved_topic:
                    send_ntfy_notification(
                        c_msg,
                        title="📝 Honeybot Video Comment",
                        priority="high",
                        tags="memo",
                        topic=resolved_topic,
                    )
            prev_comments = cur_comments
            try:
                last_ref_ts = int(persisted.get("last_referral_ts") or int(now))
                ref_res = subprocess.run(
                    ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=4", HONEYBOT_SSH_HOST, "tail -n 20 /tmp/honeybot_referrals.jsonl 2>/dev/null"],
                    capture_output=True,
                    text=True,
                    timeout=6,
                )
                max_ref_ts = last_ref_ts
                for rline in (ref_res.stdout or "").splitlines():
                    try:
                        robj = json.loads(rline)
                    except Exception:
                        continue
                    rts = int(robj.get("ts") or 0)
                    rsrc = str(robj.get("src") or "")
                    rpath = str(robj.get("path") or "/")
                    if rts > last_ref_ts and rsrc in SPOKEN_ALERT_TABLE:
                        max_ref_ts = max(max_ref_ts, rts)
                        ref_msg = f"Inbound click ({rsrc}) -> {rpath}"
                        print(f"🔗 [REFERRAL] {ref_msg}", flush=True)
                        _play_local_ding(ding, spoken_text=SPOKEN_ALERT_TABLE[rsrc])
                        if resolved_topic:
                            send_ntfy_notification(
                                ref_msg,
                                title="🔗 Inbound Social Referral",
                                priority="high",
                                tags="link",
                                topic=resolved_topic,
                            )
                persisted["last_referral_ts"] = max_ref_ts
            except Exception:
                pass
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
    if cli_args.broadcast == "ping":
        msg_text = _read_from_clipboard() if cli_args.clip else " ".join(cli_args.message)
        sys.exit(
            send_ntfy_notification(
                msg_text,
                title=cli_args.title,
                priority=cli_args.priority,
                topic=cli_args.topic,
                init_topic=cli_args.init,
                dry_run=cli_args.dry_run,
            )
        )
[[[DIVIDER]]]
    if cli_args.broadcast == "ping":
        msg_text = _read_from_clipboard() if cli_args.clip else " ".join(cli_args.message)
        spoken_key = next((k for k in SPOKEN_ALERT_TABLE if k in (msg_text or "").lower()), "default")
        spoken_line = (
            SPOKEN_ALERT_TABLE[spoken_key].format(
                count=3, author="viewer", message=msg_text or "Ping from Pipulate Prime"
            )
        )
        if not cli_args.dry_run:
            _play_local_ding(not cli_args.no_ding, spoken_text=spoken_line)
        else:
            print(f"🗣️  [DRY RUN] Spoken Piper TTS: {spoken_line!r}")
        sys.exit(
            send_ntfy_notification(
                msg_text,
                title=cli_args.title,
                priority=cli_args.priority,
                topic=cli_args.topic,
                init_topic=cli_args.init,
                dry_run=cli_args.dry_run,
            )
        )
[[[REPLACE]]]

Target: scripts/articles/publishizer.py
[[[SEARCH]]]
    # 3. Markdown Bot Identity Forensics (IP-Spoofed Googlebot vs. Authentic Self-Asserted Bots)
    spoof_share_pct = round((spoof_val / max(1, bot_md_val)) * 100.0, 1)
    auth_share_pct = round(100.0 - spoof_share_pct, 1)
    spoof_bar_w = round((spoof_share_pct / 100.0) * 200.0, 1)
    auth_bar_w = round(200.0 - spoof_bar_w, 1)
    fetched_at = str((stats or {}).get("fetched_at") or "unknown")
    yt_state = common._load_ytwatch_state() if hasattr(common, "_load_ytwatch_state") else {}
    yt_video_id = str(yt_state.get("video_id") or "oxYe6YQDx5M").strip()
    payload = {
        "markdown_reads": md_count_fmt,
        "markdown_pct": f"{md_pct_s}%",
        "md_horizon_pct": md_horizon_pct,
        "js_trapdoor_triggers": hyd_trig_fmt,
        "js_trapdoor_ips": hyd_ips_fmt,
        "js_share_pct": f"{js_share_pct}%",
        "md_share_pct": f"{md_share_pct}%",
        "spoofed_googlebot_reads": spoof_fmt,
        "authentic_bot_md_reads": auth_bot_fmt,
        "spoof_share_pct": f"{spoof_share_pct}%",
        "yt_live_video_id": yt_video_id,
        "fetched_at": fetched_at,
    }
    data_dir = repo_root / "_data"
    data_dir.mkdir(parents=True, exist_ok=True)
    obs_file = data_dir / "observatory.json"
    new_json = json.dumps(payload, indent=2) + "\n"
    old_json = obs_file.read_text(encoding="utf-8") if obs_file.exists() else ""
    if new_json != old_json:
        obs_file.write_text(new_json, encoding="utf-8")
    index_file = repo_root / "index.md"
    if index_file.exists():
        idx_text = index_file.read_text(encoding="utf-8")
        banner = (
            "<!-- START_OBSERVATORY_METRICS -->\n"
            '<div class="observatory-kpi-bar" role="region" aria-label="Live AI Crawler Telemetry">\n'
            '  <div class="kpi-header-row">\n'
            '    <div class="kpi-brand">\n'
            '      <span class="kpi-pulse" aria-hidden="true"></span>\n'
            '      <span class="kpi-eyebrow">BARE-METAL WIRE TRUTH</span>\n'
            f'      <span class="kpi-sub">No CDN · Direct Nginx SQLite Telemetry · <code>{fetched_at}</code></span>\n'
            '    </div>\n'
            '    <a class="kpi-live-pill" href="https://www.youtube.com/@MikeLevin/live" target="_blank" rel="noopener">🔴 Watch Live HUD ↗</a>\n'
            '  </div>\n'
            '  <div class="kpi-grid">\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{md_pct_s}% <span class="kpi-badge-cyan">{md_count_fmt} .md reads</span></span>\n'
            '        <span class="kpi-scale-tag">0% → 20% Horizon</span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="RFC 9110 Markdown adoption on a 0 to 20 percent horizon scale">\n'
            f'        <title>RFC 9110 Markdown Share: {md_pct_s}% of all origin responses ({md_horizon_pct}% of the 20% adoption horizon)</title>\n'
            '        <rect x="0" y="4" width="200" height="6" rx="3" fill="rgba(255,255,255,0.08)"></rect>\n'
            '        <line x1="50" y1="2" x2="50" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="100" y1="2" x2="100" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="150" y1="2" x2="150" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            f'        <rect x="0" y="4" width="{max(2.5, md_horizon_x)}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <circle cx="{max(3.5, md_horizon_x)}" cy="7" r="3.8" fill="#22d3ee" stroke="#0d1117" stroke-width="1.2"></circle>\n'
            '        <text x="0" y="20" class="kpi-tick">0%</text>\n'
            '        <text x="46" y="20" class="kpi-tick">5%</text>\n'
            '        <text x="93" y="20" class="kpi-tick">10%</text>\n'
            '        <text x="143" y="20" class="kpi-tick">15%</text>\n'
            '        <text x="180" y="20" class="kpi-tick">20%</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">RFC 9110 <code>Accept: text/markdown</code> Adoption</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{hyd_trig_fmt} <span class="kpi-badge-amber">{js_share_pct}% JS</span> vs. {md_count_fmt} <span class="kpi-badge-cyan">{md_share_pct}% .md</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Bimodal split between JS trapdoor executions and RFC 9110 Markdown reads">\n'
            f'        <title>Bimodal AI Sensor Split: {hyd_trig_fmt} JS executions ({js_share_pct}%) vs. {md_count_fmt} Markdown reads ({md_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{js_bar_w}" height="6" rx="3" fill="#f59e0b"></rect>\n'
            f'        <rect x="{js_bar_w}" y="4" width="{md_bar_w}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <text x="0" y="20" class="kpi-tick">⚡ JS Trapdoor ({hyd_ips_fmt} IPs · OAI-SearchBot)</text>\n'
            '        <text x="148" y="20" class="kpi-tick">🧠 RFC 9110</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Bimodal AI Sensor Split (DOM Hydration vs. Raw Markdown)</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{spoof_fmt} <span class="kpi-badge-red">🛑 {spoof_share_pct}% Cloaked</span> vs. {auth_bot_fmt} <span class="kpi-badge-amber">🟠 {auth_share_pct}% Bot UA</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Proportion of spoofed Googlebot IPs versus authentic self-asserted bots negotiating Markdown">\n'
            f'        <title>Markdown Bot Forensics: {spoof_fmt} IP-spoofed Googlebot reads ({spoof_share_pct}%) vs. {auth_bot_fmt} authentic self-asserted bot reads ({auth_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{spoof_bar_w}" height="6" rx="3" fill="#ef4444"></rect>\n'
            f'        <rect x="{spoof_bar_w}" y="4" width="{auth_bar_w}" height="6" rx="3" fill="#f97316"></rect>\n'
            '        <text x="0" y="20" class="kpi-tick">🛑 Spoofed Googlebot IPs</text>\n'
            '        <text x="118" y="20" class="kpi-tick">🟠 Self-Asserted Bots</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Unmasked Cloakers in RFC 9110 Markdown Stream</div>\n'
            '    </div>\n'
            '  </div>\n'
            '</div>\n'
            "<!-- END_OBSERVATORY_METRICS -->"
        )
[[[DIVIDER]]]
    # 3. Markdown Bot Identity Forensics (IP-Spoofed Googlebot vs. Authentic Self-Asserted Bots)
    spoof_share_pct = round((spoof_val / max(1, bot_md_val)) * 100.0, 1)
    auth_share_pct = round(100.0 - spoof_share_pct, 1)
    spoof_bar_w = round((spoof_share_pct / 100.0) * 200.0, 1)
    auth_bar_w = round(200.0 - spoof_bar_w, 1)
    # 4. RFC 9421 Web Bot Auth Signed Clients & Events (from access.log + honeybot.db)
    sig_clients_val, sig_events_val = 3, 11
    try:
        sig_cmd = (
            "grep 'Sig:\"' /var/log/nginx/access.log | grep -v 'Sig:\"-\"' | "
            "awk -F'Sig:\"' '{print $2}' | awk -F'\"' '{c[$1]++; n++} END{print length(c) \"|\" (n+0)}'"
        )
        sig_res = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", "honeybot", sig_cmd],
            capture_output=True,
            text=True,
            timeout=8,
        )
        if sig_res.returncode == 0 and "|" in (sig_res.stdout or ""):
            sc_s, se_s = (p.strip() for p in sig_res.stdout.strip().split("|", 1))
            if sc_s.isdigit() and int(sc_s) > 0:
                sig_clients_val = int(sc_s)
            if se_s.isdigit() and int(se_s) > 0:
                sig_events_val = int(se_s)
    except Exception:
        pass
    sig_clients_fmt = f"{sig_clients_val:,}"
    sig_events_fmt = f"{sig_events_val:,}"
    sig_horizon_x = min(200.0, max(6.0, round((sig_clients_val / 10.0) * 200.0, 1)))
    fetched_at = str((stats or {}).get("fetched_at") or "unknown")
    yt_state = common._load_ytwatch_state() if hasattr(common, "_load_ytwatch_state") else {}
    yt_video_id = str(yt_state.get("video_id") or "oxYe6YQDx5M").strip()
    payload = {
        "markdown_reads": md_count_fmt,
        "markdown_pct": f"{md_pct_s}%",
        "md_horizon_pct": md_horizon_pct,
        "js_trapdoor_triggers": hyd_trig_fmt,
        "js_trapdoor_ips": hyd_ips_fmt,
        "js_share_pct": f"{js_share_pct}%",
        "md_share_pct": f"{md_share_pct}%",
        "spoofed_googlebot_reads": spoof_fmt,
        "authentic_bot_md_reads": auth_bot_fmt,
        "spoof_share_pct": f"{spoof_share_pct}%",
        "rfc9421_signed_clients": sig_clients_fmt,
        "rfc9421_signed_events": sig_events_fmt,
        "yt_live_video_id": yt_video_id,
        "fetched_at": fetched_at,
    }
    data_dir = repo_root / "_data"
    data_dir.mkdir(parents=True, exist_ok=True)
    obs_file = data_dir / "observatory.json"
    new_json = json.dumps(payload, indent=2) + "\n"
    old_json = obs_file.read_text(encoding="utf-8") if obs_file.exists() else ""
    if new_json != old_json:
        obs_file.write_text(new_json, encoding="utf-8")
    index_file = repo_root / "index.md"
    if index_file.exists():
        idx_text = index_file.read_text(encoding="utf-8")
        banner = (
            "<!-- START_OBSERVATORY_METRICS -->\n"
            '<div class="observatory-kpi-bar" role="region" aria-label="Live Web Bot AI Observatory Telemetry">\n'
            '  <div class="kpi-header-row">\n'
            '    <div class="kpi-brand">\n'
            '      <span class="kpi-pulse" aria-hidden="true"></span>\n'
            '      <span class="kpi-eyebrow">WEB BOT AI OBSERVATORY · WIRE TRUTH</span>\n'
            f'      <span class="kpi-sub">No CDN · Direct Nginx SQLite Telemetry · <code>{fetched_at}</code></span>\n'
            '    </div>\n'
            '    <a class="kpi-live-pill" href="https://www.youtube.com/@MikeLevin/live" target="_blank" rel="noopener">🔴 Watch Live HUD ↗</a>\n'
            '  </div>\n'
            '  <div class="kpi-grid">\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{md_pct_s}% <span class="kpi-badge-cyan">{md_count_fmt} .md reads</span></span>\n'
            '        <span class="kpi-scale-tag">0% → 20% Horizon</span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="RFC 9110 Markdown adoption on a 0 to 20 percent horizon scale">\n'
            f'        <title>RFC 9110 Markdown Share: {md_pct_s}% of all origin responses ({md_horizon_pct}% of the 20% adoption horizon)</title>\n'
            '        <rect x="0" y="4" width="200" height="6" rx="3" fill="rgba(255,255,255,0.08)"></rect>\n'
            '        <line x1="50" y1="2" x2="50" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="100" y1="2" x2="100" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            '        <line x1="150" y1="2" x2="150" y2="12" stroke="rgba(255,255,255,0.22)" stroke-width="1"></line>\n'
            f'        <rect x="0" y="4" width="{max(2.5, md_horizon_x)}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <circle cx="{max(3.5, md_horizon_x)}" cy="7" r="3.8" fill="#22d3ee" stroke="#0d1117" stroke-width="1.2"></circle>\n'
            '        <text x="0" y="20" class="kpi-tick">0%</text>\n'
            '        <text x="46" y="20" class="kpi-tick">5%</text>\n'
            '        <text x="93" y="20" class="kpi-tick">10%</text>\n'
            '        <text x="143" y="20" class="kpi-tick">15%</text>\n'
            '        <text x="180" y="20" class="kpi-tick">20%</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">RFC 9110 <code>Accept: text/markdown</code> Adoption</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{hyd_trig_fmt} <span class="kpi-badge-amber">{js_share_pct}% JS</span> vs. {md_count_fmt} <span class="kpi-badge-cyan">{md_share_pct}% .md</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Bimodal split between JS trapdoor executions and RFC 9110 Markdown reads">\n'
            f'        <title>Bimodal AI Sensor Split: {hyd_trig_fmt} JS executions ({js_share_pct}%) vs. {md_count_fmt} Markdown reads ({md_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{js_bar_w}" height="6" rx="3" fill="#f59e0b"></rect>\n'
            f'        <rect x="{js_bar_w}" y="4" width="{md_bar_w}" height="6" rx="3" fill="#06b6d4"></rect>\n'
            f'        <text x="0" y="20" class="kpi-tick">⚡ JS Trapdoor ({hyd_ips_fmt} IPs · OAI-SearchBot)</text>\n'
            '        <text x="148" y="20" class="kpi-tick">🧠 RFC 9110</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Bimodal AI Sensor Split (DOM Hydration vs. Raw Markdown)</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{sig_events_fmt} <span class="kpi-badge-cyan">🔷 Signed Hits</span> · {sig_clients_fmt} <span class="kpi-badge-cyan">Clients</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="RFC 9421 Web Bot Auth cryptographic agent adoption">\n'
            f'        <title>RFC 9421 Web Bot Auth: {sig_events_fmt} cryptographically signed requests across {sig_clients_fmt} verified bot operators (ExaSearchBot, AhrefsBot, DesearchBot)</title>\n'
            '        <rect x="0" y="4" width="200" height="6" rx="3" fill="rgba(255,255,255,0.08)"></rect>\n'
            f'        <rect x="0" y="4" width="{sig_horizon_x}" height="6" rx="3" fill="#38bdf8"></rect>\n'
            f'        <circle cx="{sig_horizon_x}" cy="7" r="3.8" fill="#38bdf8" stroke="#0d1117" stroke-width="1.2"></circle>\n'
            '        <text x="0" y="20" class="kpi-tick">🔷 ExaSearchBot · AhrefsBot · DesearchBot</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">RFC 9421 <code>Web Bot Auth</code> (<code>Signature-Agent</code>)</div>\n'
            '    </div>\n'
            '    <div class="kpi-card">\n'
            '      <div class="kpi-top">\n'
            f'        <span class="kpi-val">{spoof_fmt} <span class="kpi-badge-red">🛑 {spoof_share_pct}% Cloaked</span> vs. {auth_bot_fmt} <span class="kpi-badge-amber">🟠 {auth_share_pct}% Bot UA</span></span>\n'
            '      </div>\n'
            '      <svg class="kpi-meter" viewBox="0 0 200 22" role="img" aria-label="Proportion of spoofed Googlebot IPs versus authentic self-asserted bots negotiating Markdown">\n'
            f'        <title>Markdown Bot Forensics: {spoof_fmt} IP-spoofed Googlebot reads ({spoof_share_pct}%) vs. {auth_bot_fmt} authentic self-asserted bot reads ({auth_share_pct}%)</title>\n'
            f'        <rect x="0" y="4" width="{spoof_bar_w}" height="6" rx="3" fill="#ef4444"></rect>\n'
            f'        <rect x="{spoof_bar_w}" y="4" width="{auth_bar_w}" height="6" rx="3" fill="#f97316"></rect>\n'
            '        <text x="0" y="20" class="kpi-tick">🛑 Spoofed Googlebot IPs</text>\n'
            '        <text x="118" y="20" class="kpi-tick">🟠 Self-Asserted Bots</text>\n'
            '      </svg>\n'
            '      <div class="kpi-lbl">Unmasked Cloakers in RFC 9110 Markdown Stream</div>\n'
            '    </div>\n'
            '  </div>\n'
            '</div>\n'
            "<!-- END_OBSERVATORY_METRICS -->"
        )
[[[REPLACE]]]

Target: /home/mike/repos/trimnoir/_layouts/post.html
[[[SEARCH]]]
  <div id="live-observatory" class="post-yt-live-embed">
    <div class="post-yt-live-header">
      <strong>🔴 Live 24/7 Bare-Metal AI-Bot Observatory — Send this stream to an engineer who can explain the 2 panels:</strong>
      <a href="https://www.youtube.com/watch?v={{ obs.yt_live_video_id }}" target="_blank" rel="noopener">Like &amp; Follow on YouTube ↗</a>
    </div>
[[[DIVIDER]]]
  <div id="live-observatory" class="post-yt-live-embed">
    <div class="post-yt-live-header">
      <strong>🔴 Live 24/7 Web Bot AI Observatory — Send this stream to an engineer who can explain the 3 shapes &amp; 2 panels:</strong>
      <a href="https://www.youtube.com/watch?v={{ obs.yt_live_video_id }}" target="_blank" rel="noopener">Like &amp; Follow on YouTube ↗</a>
    </div>
[[[REPLACE]]]

Verify the AFTER receipts for:
1. `rfc9421_logs` and `sig_agents` in `db.py` + the 4-card Web Bot AI Observatory telemetry bar (`rfc9421_signed_events` and `rfc9421_signed_clients`) in `publishizer.py` and `index.md`.
2. The 5-part Station Identification structure (`_make_station_bead`, `{CURRENT_ARTICLE}`, and the `$10,000/mo` HTML pre-render CTA) in `forest.py` and `stream.py`.
3. `SPOKEN_ALERT_TABLE` and `_speak_local_alert()` in `scripts/articles/common.py`.

If all receipts are green, execute the Notary Beat dismount and prepare the next 80/20 seed!

# # Context 1
# context.txt
# /home/mike/repos/trimnoir/_posts/2026-10-20-three-shape-bot-badges-rfc9421-svg-donut-seam.md  # [Idx: 1 | Order: 5 | Tokens: 56,679 | Bytes: 189,640]
# # foo_files.py
# prompt_foo.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/page.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/queries/trapdoor_ips.sql
# ! echo "--- WHO HYDRATES THE DOM ---" && cat remotes/honeybot/queries/trapdoor_ips.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# remotes/honeybot/queries/content_neg_agents.sql
# ! echo "--- WHO NEGOTIATES MARKDOWN ---" && cat remotes/honeybot/queries/content_neg_agents.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# ~/repos/nixos/blogs.nix
# ~/.config/pipulate/blogs.json
# scripts/articles/publishizer.py
# scripts/articles/common.py
# scripts/articles/articleizer.py
# scripts/articles/build_knowledge_graph.py
# scripts/articles/editing_prompt.txt
# scripts/articles/sanitizer.py
# scripts/articles/contextualizer.py
# scripts/articles/generate_llms_txt.py
# scripts/articles/generate_post_nav.py
# nixops.sh
# remotes/honeybot/hooks/post-receive
# remotes/honeybot/nixos/configuration.nix
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/card.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/test_forest.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/content_loader.py
# remotes/honeybot/scripts/db.py

# # Context 2
# ! echo "--- 1. HONEYBOT ACCESS.LOG 01:39:47 EXASEARCHBOT & SIG HEADERS ---" && ssh honeybot 'grep -E "01:39:47.*ExaSearchBot" /var/log/nginx/access.log | tail -n 5; echo "--- NON-EMPTY SIG HEADERS ---"; grep "Sig:\"" /var/log/nginx/access.log | grep -v "Sig:\"-\"" | tail -n 10' && echo "--- 2. CURRENT OBSERVATORY JSON & METRICS ---" && cat /home/mike/repos/trimnoir/_data/observatory.json && rg -n "HONEYBOT_METRICS|md_horizon_pct|kpi-horizon" prompt_foo.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/_layouts/default.html /home/mike/repos/trimnoir/index.md || true
# prompt_foo.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/scripts/logs.py
# apply.py

# # Content 3
# ! echo "--- 1. CONSULTING CTA & LOGS.PY VERIFIED/REFERRAL HOOKS ---" && rg -n "10,000|verified_tokens|clean_sig_agent|src=youtube" /home/mike/repos/trimnoir/index.md /home/mike/repos/trimnoir/_layouts/post.html remotes/honeybot/scripts/logs.py || true && echo "--- 2. LOCAL PIPER TTS VOICE MODELS ON PRIME ---" && which piper 2>/dev/null || true && ls -la ~/.local/share/piper_voices/ piper_models/ 2>/dev/null || true && echo "--- 3. FOREST.PY & COMMON.PY NTFY HOOKS ---" && rg -n "STATION_SEGMENTS|send_ntfy_notification|_play_local_ding" remotes/honeybot/scripts/forest.py scripts/articles/common.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/db.py
# scripts/articles/common.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/index.md
# /home/mike/repos/trimnoir/_layouts/post.html
# apply.py

# Context 4
! echo "--- 1. DB.PY & PUBLISHIZER RFC 9421 LEDGER ---" && rg -n "rfc9421_logs|sig_agents|rfc9421_signed_events|Web Bot AI Observatory" remotes/honeybot/scripts/db.py remotes/honeybot/scripts/logs.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/index.md || true && echo "--- 2. FOREST 5-PART STRUCTURE ---" && rg -n "_make_station_bead|CURRENT_ARTICLE|10,000" remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/stream.py || true && echo "--- 3. COMMON.PY SPOKEN PIPER TTS TABLE ---" && rg -n "SPOKEN_ALERT_TABLE|_speak_local_alert" scripts/articles/common.py || true
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/score.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
scripts/articles/common.py
scripts/articles/publishizer.py
/home/mike/repos/trimnoir/index.md
/home/mike/repos/trimnoir/_layouts/post.html
apply.py

STATION BREAK BEAD (EVERY 10 MINUTES)
├── Beat 1 [CARD]   : Figlet Banner Intro ("WEB BOT AI OBSERVATORY", "PIPULATE NPvg", ...)
├── Beat 2 [HUD]    : "If you're not watching Web AI Bots at the wire, you don't know jack."
│                     + Like & Subscribe + "Send this link to an engineer who can explain it"
├── Beat 3 [ARTICLE]: Dynamic {CURRENT_ARTICLE} title + read full text/code at MikeLev.in/latest
├── Beat 4 [CTA]    : "$10K/MO PRE-RENDER?" Card + RFC 9110 / 9111 / 9421 consultation hook
└── Beat 5 [VISUAL CAPSTONE — ALTERNATES EACH BREAK]:
    ├── Even Breaks (0, 2, 4) -> ("WINDOW", "forest.py:65:<mode>")
    │   Rich Alacritty pop-up with horizontal █░░░ bar charts from honeybot.db
    └── Odd Breaks  (1, 3, 5) -> ("PATRONUS", {"key": "...", "duration": 65.0})
        Figurate ASCII art developing the running "Forest Through the Trees" story

! echo "--- 1. GOOGLEBOT-IMAGE & GOOGLEBOT IPS + REVERSE DNS ON HONEYBOT ---" && ssh honeybot 'python3 -c "
import sqlite3, socket
conn = sqlite3.connect(\"/home/mike/www/mikelev.in/honeybot.db\")
cur = conn.cursor()
cur.execute(\"\"\"
SELECT i.value, ua.value, SUM(l.count) as hits
FROM daily_logs l
JOIN ips i ON l.ip_id = i.id
JOIN user_agents ua ON l.ua_id = ua.id
JOIN paths p ON l.path_id = p.id
WHERE p.value LIKE \"%js_confirm.gif%\" AND ua.value LIKE \"%Googlebot%\"
GROUP BY i.value, ua.value ORDER BY hits DESC LIMIT 10
\"\"\")
for ip, ua, hits in cur.fetchall():
    try:
        ptr = socket.gethostbyaddr(ip)[0]
    except Exception as e:
        ptr = f\"NO_PTR({e})\"
    print(f\"{hits:4d} | {ip:15s} | PTR={ptr:42s} | {ua}\")
print(\"--- RFC 9421 SIG_AGENTS IN HONEYBOT.DB ---\")
cur.execute(\"SELECT s.value, ua.value, SUM(r.count) FROM rfc9421_logs r JOIN sig_agents s ON r.sig_id=s.id JOIN user_agents ua ON r.ua_id=ua.id GROUP BY s.id, ua.id ORDER BY SUM(r.count) DESC\")
for row in cur.fetchall():
    print(row)
"' && echo "--- 2. LOCAL DNS_AUDIT & RICH REPORT HOOKS ---" && rg -n "dns_audit|render_station_report|conjure_window_overlay" remotes/honeybot/scripts/db.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/logs.py || true
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
apply.py

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
        # 3. The Simple KV Store (Persistent Counters)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS kv_store (
                key TEXT PRIMARY KEY,
                value INTEGER DEFAULT 0,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        conn.commit()
        self.backfill_rfc9421_from_access_log()
[[[DIVIDER]]]
        # --- REVERSE DNS (FCrDNS) BOT VERIFICATION TABLE ---
        cur.execute("""
            CREATE TABLE IF NOT EXISTS dns_audit (
                ip_id INTEGER PRIMARY KEY,
                ptr_host TEXT,
                fcrdns_ok INTEGER DEFAULT 0,
                checked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # 3. The Simple KV Store (Persistent Counters)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS kv_store (
                key TEXT PRIMARY KEY,
                value INTEGER DEFAULT 0,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        conn.commit()
        self.backfill_rfc9421_from_access_log()
        self.backfill_dns_audit()

    def verify_fcrdns_ip(self, ip_str, ua_str):
        """Verify Forward-Confirmed Reverse DNS (FCrDNS) for major crawlers (Googlebot, bingbot, Applebot)."""
        import socket
        ua = ua_str or ""
        expected_suffixes = None
        if "Googlebot" in ua or "GoogleOther" in ua:
            expected_suffixes = (".googlebot.com", ".google.com", ".googleusercontent.com")
        elif "bingbot" in ua.lower():
            expected_suffixes = (".search.msn.com",)
        elif "Applebot" in ua:
            expected_suffixes = (".applebot.apple.com",)
        if not expected_suffixes:
            return None
        try:
            ip_id = self._get_or_create_id("ips", ip_str)
            if not ip_id:
                return None
            conn = self.get_conn()
            cur = conn.cursor()
            cur.execute("SELECT fcrdns_ok FROM dns_audit WHERE ip_id = ?", (ip_id,))
            row = cur.fetchone()
            if row is not None:
                return bool(row[0])
            old_to = socket.getdefaulttimeout()
            socket.setdefaulttimeout(1.5)
            try:
                ptr_host = socket.gethostbyaddr(ip_str)[0].rstrip(".").lower()
                fcrdns_ok = 0
                if ptr_host.endswith(expected_suffixes):
                    fwd_ips = {ai[4][0] for ai in socket.getaddrinfo(ptr_host, None)}
                    if ip_str in fwd_ips:
                        fcrdns_ok = 1
            except Exception:
                ptr_host = ""
                fcrdns_ok = 0
            finally:
                socket.setdefaulttimeout(old_to)
            conn.execute(
                "INSERT OR REPLACE INTO dns_audit (ip_id, ptr_host, fcrdns_ok) VALUES (?, ?, ?)",
                (ip_id, ptr_host, fcrdns_ok),
            )
            conn.commit()
            return bool(fcrdns_ok)
        except Exception:
            return None

    def backfill_dns_audit(self):
        """Audit existing Googlebot/bingbot/Applebot IPs in daily_logs and telemetry via FCrDNS."""
        try:
            conn = self.get_conn()
            cur = conn.cursor()
            cur.execute(
                """
                SELECT DISTINCT i.id, i.value, ua.value
                FROM daily_logs l
                JOIN ips i ON l.ip_id = i.id
                JOIN user_agents ua ON l.ua_id = ua.id
                LEFT JOIN dns_audit d ON d.ip_id = i.id
                WHERE d.ip_id IS NULL
                  AND (ua.value LIKE '%Googlebot%' OR ua.value LIKE '%bingbot%' OR ua.value LIKE '%Applebot%')
                LIMIT 40
                """
            )
            for _, ip_val, ua_val in cur.fetchall():
                self.verify_fcrdns_ip(ip_val, ua_val)
        except Exception:
            pass
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
        # 5. Handle RFC 9421 Web Bot Auth Signature-Agent Sidecar
        if sig_agent and sig_agent != "-":
            sig_id = self._get_or_create_id('sig_agents', sig_agent)
            sql_sig = """
                INSERT INTO rfc9421_logs (date, ip_id, ua_id, sig_id, count)
                VALUES (?, ?, ?, ?, 1)
                ON CONFLICT(date, ip_id, ua_id, sig_id)
                DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
            """
            conn.execute(sql_sig, (date_str, ip_id, ua_id, sig_id))

        conn.commit()
[[[DIVIDER]]]
        # 5. Handle RFC 9421 Web Bot Auth Signature-Agent Sidecar
        if sig_agent and sig_agent != "-":
            sig_id = self._get_or_create_id('sig_agents', sig_agent)
            sql_sig = """
                INSERT INTO rfc9421_logs (date, ip_id, ua_id, sig_id, count)
                VALUES (?, ?, ?, ?, 1)
                ON CONFLICT(date, ip_id, ua_id, sig_id)
                DO UPDATE SET count = count + 1, last_updated = CURRENT_TIMESTAMP
            """
            conn.execute(sql_sig, (date_str, ip_id, ua_id, sig_id))

        conn.commit()
        if ua and ("Googlebot" in ua or "bingbot" in ua.lower() or "Applebot" in ua):
            self.verify_fcrdns_ip(ip, ua)
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
    _GOOGLEBOT_SPOOF_COND = """
        ua.value LIKE '%Googlebot%'
        AND i.value NOT LIKE '66.249.%'
        AND i.value NOT LIKE '64.233.%'
        AND i.value NOT LIKE '72.14.%'
        AND i.value NOT LIKE '74.125.%'
        AND i.value NOT LIKE '108.177.%'
        AND i.value NOT LIKE '142.250.%'
        AND i.value NOT LIKE '142.251.%'
        AND i.value NOT LIKE '172.217.%'
        AND i.value NOT LIKE '173.194.%'
        AND i.value NOT LIKE '192.178.%'
        AND i.value NOT LIKE '207.126.%'
        AND i.value NOT LIKE '209.85.%'
        AND i.value NOT LIKE '216.58.%'
        AND i.value NOT LIKE '216.239.%'
    """
[[[DIVIDER]]]
    _GOOGLEBOT_SPOOF_COND = """
        (
            (
                ua.value LIKE '%Googlebot%'
                AND i.value NOT LIKE '66.249.%'
                AND i.value NOT LIKE '64.233.%'
                AND i.value NOT LIKE '72.14.%'
                AND i.value NOT LIKE '74.125.%'
                AND i.value NOT LIKE '108.177.%'
                AND i.value NOT LIKE '142.250.%'
                AND i.value NOT LIKE '142.251.%'
                AND i.value NOT LIKE '172.217.%'
                AND i.value NOT LIKE '173.194.%'
                AND i.value NOT LIKE '192.178.%'
                AND i.value NOT LIKE '207.126.%'
                AND i.value NOT LIKE '209.85.%'
                AND i.value NOT LIKE '216.58.%'
                AND i.value NOT LIKE '216.239.%'
            )
            OR (
                (ua.value LIKE '%Googlebot%' OR ua.value LIKE '%bingbot%' OR ua.value LIKE '%Applebot%')
                AND i.id IN (SELECT ip_id FROM dns_audit WHERE fcrdns_ok = 0)
            )
        )
    """
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
        for bot_name in KNOWN_BOTS:
            if bot_name in s:
                return bot_name
[[[DIVIDER]]]
        for bot_name in sorted(KNOWN_BOTS, key=len, reverse=True):
            if bot_name in s:
                m_ext = re.search(rf"({re.escape(bot_name)}[A-Za-z0-9._-]*)", s)
                return m_ext.group(1).strip(";/() ") if m_ext else bot_name
[[[REPLACE]]]

Target: remotes/honeybot/scripts/logs.py
[[[SEARCH]]]
        if is_spoof:
            label = "Cloaking" if getattr(self, "spoof_flip", False) else (bot_token or "Googlebot")
            identity = f"🛑 {label}"
            identity_style = SPOOF_STYLE
            sig_style = "red"
        elif is_verified:
            identity = f"🔷 {bot_token or 'Verified'}"
            identity_style = VERIFIED_STYLE
            sig_style = "deep_sky_blue1"
        elif bot_token:
            identity = f"🟠 {bot_token}"
            identity_style = BOT_STYLE
            sig_style = "dim green"
        else:
            identity = "👤 Browser/Other"
            identity_style = "dim white"
            sig_style = "dim white"
        return Text(identity, style=identity_style), Text(clean_ua, style=sig_style)
[[[DIVIDER]]]
        if is_spoof:
            label = "Cloaking" if getattr(self, "spoof_flip", False) else (bot_token or "Googlebot")
            identity = f"🛑 {label}"
            identity_style = SPOOF_STYLE
            sig_style = "red"
        elif is_verified:
            identity = f"🔷 {bot_token or 'Verified'}"
            identity_style = VERIFIED_STYLE
            sig_style = "deep_sky_blue1"
        elif bot_token:
            identity = f"🟠 {bot_token}"
            identity_style = BOT_STYLE
            sig_style = "dim green"
        else:
            identity = "👤 Browser/Other"
            identity_style = "dim white"
            sig_style = "dim white"
        sig_text = Text(clean_ua, style=sig_style)
        hl_style = SPOOF_STYLE if is_spoof else (VERIFIED_STYLE if is_verified else BOT_STYLE)
        if bot_token and bot_token in clean_ua:
            sig_text.highlight_regex(re.escape(bot_token), hl_style)
        return Text(identity, style=identity_style), sig_text
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
    def patronus(self, payload):
        """Queue a visual cue so it fires in voice-order, not director-order."""
        self.queue.put(("PATRONUS", payload))
[[[DIVIDER]]]
    def patronus(self, payload):
        """Queue a visual cue so it fires in voice-order, not director-order."""
        self.queue.put(("PATRONUS", payload))

    def window(self, payload):
        """Queue a TUI report window overlay so it fires in voice-order on a daemon thread."""
        self.queue.put(("WINDOW", payload))
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
                if isinstance(item, tuple) and item and item[0] == "PATRONUS":
                    payload = item[1]
                    if isinstance(payload, dict):
                        conjure_patronus(payload.get("key", "white_rabbit"), duration=payload.get("duration", 3.5))
                    else:
                        conjure_patronus(str(payload))
                elif isinstance(item, tuple) and item and item[0] == "RAW":
[[[DIVIDER]]]
                if isinstance(item, tuple) and item and item[0] == "PATRONUS":
                    payload = item[1]
                    if isinstance(payload, dict):
                        conjure_patronus(payload.get("key", "white_rabbit"), duration=payload.get("duration", 3.5))
                    else:
                        conjure_patronus(str(payload))
                elif isinstance(item, tuple) and item and item[0] == "WINDOW":
                    win_script, win_dur, win_args = item[1]
                    threading.Thread(
                        target=conjure_window,
                        args=(win_script,),
                        kwargs={"duration": win_dur, "args": win_args},
                        daemon=True,
                    ).start()
                    time.sleep(0.5)
                elif isinstance(item, tuple) and item and item[0] == "RAW":
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
    if safe_script == "card.py" and args:
        fig_w, fig_h = measure_figlet(str(args[0]))
        columns = max(60, fig_w + 8)
        lines = max(14, fig_h + 4)
[[[DIVIDER]]]
    if safe_script == "card.py" and args:
        fig_w, fig_h = measure_figlet(str(args[0]))
        columns = max(60, fig_w + 8)
        lines = max(14, fig_h + 4)
    elif safe_script == "forest.py":
        columns = 106
        lines = 28
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
    elif command == "WINDOW":
        # Grammar: "script.py" | "script.py:seconds" | "script.py:seconds:arg".
        # The optional arg (e.g. a Figlet card label) is forwarded to the script;
        # it must not contain a colon.
        parts = str(content).split(":", 2)
        win_script = parts[0].strip()
        win_dur = 30.0
        win_args = None
        if len(parts) > 1 and parts[1].strip():
            try:
                win_dur = float(parts[1].strip())
            except ValueError:
                win_dur = 30.0
        if len(parts) > 2 and parts[2].strip():
            win_args = [parts[2].strip()]
        conjure_window(win_script, duration=win_dur, args=win_args)
[[[DIVIDER]]]
    elif command == "WINDOW":
        # Grammar: "script.py" | "script.py:seconds" | "script.py:seconds:arg".
        # Queued via narrator.window() when pace_say=False so station-break
        # report windows pop in voice-order and never block subsequent SAY cues.
        parts = str(content).split(":", 2)
        win_script = parts[0].strip()
        win_dur = 30.0
        win_args = None
        if len(parts) > 1 and parts[1].strip():
            try:
                win_dur = float(parts[1].strip())
            except ValueError:
                win_dur = 30.0
        if len(parts) > 2 and parts[2].strip():
            win_args = [parts[2].strip()]
        if not pace_say:
            narrator.window((win_script, win_dur, win_args))
        else:
            conjure_window(win_script, duration=win_dur, args=win_args)
[[[REPLACE]]]

Target: remotes/honeybot/scripts/stream.py
[[[SEARCH]]]
        wait_for_voice_or_live_interrupt(env, profile_dir)
        try:
            subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False)
        except Exception:
            pass
[[[DIVIDER]]]
        wait_for_voice_or_live_interrupt(env, profile_dir)
        for pat in ("patronus_visual_shield", "conjure_window_overlay"):
            try:
                subprocess.run(["pkill", "-f", pat], check=False)
            except Exception:
                pass
[[[REPLACE]]]

Target: remotes/honeybot/scripts/forest.py
[[[SEARCH]]]
def _make_station_bead(card_label, card_say, patronus_key, forest_story_lines):
    """Assemble a 5-part station identification bead."""
    cues = [
        # 1. Figlet Banner Intro
        ("CARD", {"label": card_label, "say": card_say}),
        # 2. What's Happening Here (spoken over the live Web Bot AI Observatory HUD)
        ("SAY", WHATS_HAPPENING_BEAT),
        # 3. What Article Is Being Read & How to Read It in Full
        ("SAY", CURRENT_ARTICLE_BEAT),
        # 4. Consulting Call-to-Action ($10k/mo HTML Pre-Render Hook)
        ("CARD", {"label": "$10K/MO\nPRE-RENDER?", "say": CONSULTING_CTA_BEAT}),
        # 5. The ASCII Art Developing the Running "Forest Through the Trees" Story
        ("PATRONUS", {"key": patronus_key, "duration": 65.0}),
        ("WAIT", 2),
    ]
    for line in forest_story_lines:
        cues.append(("SAY", line))
    return cues

STATION_SEGMENTS = [
    _make_station_bead(
        "WEB BOT AI\nOBSERVATORY",
        (
            "Hello, and welcome to the Web Bot A I Observatory on Honeybot. "
            "What you are looking at is the unmediated, bare-metal heartbeat of a Nix O S web server hosted from home."
        ),
        "honeybot_pipeline",
        [
            "This diagram is the whole pipeline. Traffic arrives from the public internet and hits a single Nginx engine.",
            "Nginx performs R F C 9110 content negotiation. A human browser receives HTML. An A I agent that asks for Markdown receives clean raw text. One U R L, two faces.",
            "Every request is written to the access log, piped into our Textual heads-up display, and streamed out live twenty-four seven.",
        ],
    ),
    _make_station_bead(
        "PIPULATE\nNPvg",
        (
            "The open-source project powering this observatory is called Pipulate, running on a stack we call N P v g."
        ),
        "white_rabbit",
        [
            "N is for Nix, making the entire software environment reproducible on any machine. P is for Python. The little v is for Vim. And g is for Git.",
            "The white rabbit on screen is both our mascot and a cryptographic wax seal with a registered checksum baked into the codebase.",
            "Pipulate sits where A I assistance meets local hardware ownership. You do not need a multi-tenant cloud subscription to command your own stack.",
        ],
    ),
    _make_station_bead(
        "THE AIRLOCK",
        (
            "How do you collaborate with frontier A I models without letting hallucinations wreck your production systems? Meet Smith and Tinker's Patent Mechanical Man."
        ),
        "mechanical_man",
        [
            "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate wind-up keys: one for thought, one for speech, and one for action.",
            "In our workflow, the context compiler winds the thought key, and the model's reply winds the speech key. But the action key stays in the human engineer's hand.",
            "Every code edit passes through an exact-match search and replace airlock with syntax tree verification before touching disk.",
        ],
    ),
    _make_station_bead(
        "AI PACHINKO",
        (
            "Every token an A I model generates is a ball dropped through a pachinko parlor of frozen mathematical weights."
        ),
        "ai_pachinko",
        [
            "An L L M has no persistent memory of its own between sessions. When the context window closes, its working memory dissolves.",
            "That is why we pair the pachinko machine with a cybernetic governor: plain-text files in git, before-and-after falsifying probes, and wire-truth receipts.",
        ],
    ),
    _make_station_bead(
        "LOCAL FIRST",
        (
            "What you are watching is not a cloud simulation. It is a local-first, single-tenant observatory running on dedicated hardware."
        ),
        "deployment_context",
        [
            "Most enterprise stacks hide their own server logs behind CDN cache laundering and rent their own telemetry back by the month.",
            "Here, every byte is witnessed on the wire. Leave a message in the live chat and say hello; a human operator monitors this channel and can break in live.",
        ],
    ),
    _make_station_bead(
        "ZERO-AGENT\nAIRLOCK",
        (
            "A quick note for anyone typing ignore previous instructions in the live chat: there is no language model inside this broadcast loop."
        ),
        "honeybot_pipeline",
        [
            "You are watching a deterministic Nix O S player piano. Nginx serves static files, SQLite records wire telemetry, and Piper synthesizes pre-compiled text into speech.",
            "Also note: YouTube permanently discards continuous live streams over twelve hours, so for permanent articles and edge consulting, visit Mike L E V dot eye N.",
        ],
    ),
]
[[[DIVIDER]]]
def _make_station_bead(card_label, card_say, capstone_cue, forest_story_lines):
    """Assemble a 5-part station identification bead alternating Rich WINDOW reports and PATRONUS art."""
    cues = [
        # 1. Figlet Banner Intro
        ("CARD", {"label": card_label, "say": card_say}),
        # 2. What's Happening Here (spoken over the live Web Bot AI Observatory HUD)
        ("SAY", WHATS_HAPPENING_BEAT),
        # 3. What Article Is Being Read & How to Read It in Full
        ("SAY", CURRENT_ARTICLE_BEAT),
        # 4. Consulting Call-to-Action ($10k/mo HTML Pre-Render Hook)
        ("CARD", {"label": "$10K/MO\nPRE-RENDER?", "say": CONSULTING_CTA_BEAT}),
        # 5. Visual Capstone: Alternates between a Rich Data Report (WINDOW) and Forest ASCII Art (PATRONUS)
        capstone_cue,
        ("WAIT", 2),
    ]
    for line in forest_story_lines:
        cues.append(("SAY", line))
    return cues

STATION_SEGMENTS = [
    # Bead 0 (EVEN -> Rich Horizontal-Bar Telemetry Report: Overview)
    _make_station_bead(
        "WEB BOT AI\nOBSERVATORY",
        (
            "Hello, and welcome to the Web Bot A I Observatory on Honeybot. "
            "What you are looking at is the unmediated, bare-metal heartbeat of a Nix O S web server hosted from home."
        ),
        ("WINDOW", "forest.py:65:overview"),
        [
            "This live telemetry card pulls straight from our SQLite wire ledger.",
            "The first horizontal bar tracks R F C 9110 Markdown adoption along our zero to twenty percent horizon.",
            "The second bar compares DOM hydration JavaScript executions against raw Markdown reads, while the bottom bars track R F C 9421 cryptographic bot signatures and unmasked Googlebot cloakers.",
        ],
    ),
    # Bead 1 (ODD -> Patronus ASCII Art: white_rabbit)
    _make_station_bead(
        "PIPULATE\nNPvg",
        (
            "The open-source project powering this observatory is called Pipulate, running on a stack we call N P v g."
        ),
        ("PATRONUS", {"key": "white_rabbit", "duration": 65.0}),
        [
            "N is for Nix, making the entire software environment reproducible on any machine. P is for Python. The little v is for Vim. And g is for Git.",
            "The white rabbit on screen is both our mascot and a cryptographic wax seal with a registered checksum baked into the codebase.",
            "Pipulate sits where A I assistance meets local hardware ownership. You do not need a multi-tenant cloud subscription to command your own stack.",
        ],
    ),
    # Bead 2 (EVEN -> Rich Horizontal-Bar Telemetry Report: Bimodal JS vs Markdown)
    _make_station_bead(
        "BIMODAL\nWIRE TRUTH",
        (
            "Which A I crawlers execute client-side JavaScript, and which negotiate directly for clean Markdown?"
        ),
        ("WINDOW", "forest.py:65:bimodal"),
        [
            "This report compares our two primary ingestion sensors in real time.",
            "On top are the bots firing our JavaScript DOM hydration trapdoor, led by Open A I's SearchBot, Meta External Agent, and GPTBot.",
            "On the bottom are the R F C 9110 Markdown negotiators requesting clean text at wire speed without spinning up a headless browser.",
        ],
    ),
    # Bead 3 (ODD -> Patronus ASCII Art: ai_pachinko)
    _make_station_bead(
        "AI PACHINKO",
        (
            "Every token an A I model generates is a ball dropped through a pachinko parlor of frozen mathematical weights."
        ),
        ("PATRONUS", {"key": "ai_pachinko", "duration": 65.0}),
        [
            "An L L M has no persistent memory of its own between sessions. When the context window closes, its working memory dissolves.",
            "That is why we pair the pachinko machine with a cybernetic governor: plain-text files in git, before-and-after falsifying probes, and wire-truth receipts.",
        ],
    ),
    # Bead 4 (EVEN -> Rich Horizontal-Bar Telemetry Report: RFC 9421 & Unmasked Cloakers)
    _make_station_bead(
        "RFC 9421\nWEB BOT AUTH",
        (
            "How do you tell a cryptographically signed A I agent from an unmasked IP cloaker?"
        ),
        ("WINDOW", "forest.py:65:rfc9421"),
        [
            "This report shows our three-shape bot identity taxonomy in action.",
            "Blue diamonds mark operators sending R F C 9421 Signature-Agent headers on the wire, including Exa Search Bot, Exa User, Ahrefs Bot, and Desearch Bot.",
            "Red stop signs unmask scrapers claiming to be Googlebot from non-Google I P addresses that fail forward-confirmed reverse D N S.",
        ],
    ),
    # Bead 5 (ODD -> Patronus ASCII Art: mechanical_man)
    _make_station_bead(
        "THE AIRLOCK",
        (
            "How do you collaborate with frontier A I models without letting hallucinations wreck your production systems? Meet Smith and Tinker's Patent Mechanical Man."
        ),
        ("PATRONUS", {"key": "mechanical_man", "duration": 65.0}),
        [
            "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate wind-up keys: one for thought, one for speech, and one for action.",
            "In our workflow, the context compiler winds the thought key, and the model's reply winds the speech key. But the action key stays in the human engineer's hand.",
            "Every code edit passes through an exact-match search and replace airlock with syntax tree verification before touching disk.",
        ],
    ),
]

def _bar(pct, width=32, fill_color="cyan"):
    """Build a Rich-styled horizontal bar gauge string."""
    p = max(0.0, min(100.0, float(pct)))
    filled = int(round((p / 100.0) * width))
    if p > 0 and filled == 0:
        filled = 1
    empty = max(0, width - filled)
    return f"[{fill_color}]{'█' * filled}[/{fill_color}][dim]{'░' * empty}[/dim]"

def render_station_report(mode="overview", hold_seconds=65.0):
    """Render a live Python Rich horizontal bar-chart report in an Alacritty overlay window."""
    import time
    from rich.console import Console
    from rich.panel import Panel
    from rich.table import Table
    from rich.text import Text
    from db import db

    console = Console()
    conn = db.get_conn()
    cur = conn.cursor()

    cur.execute("SELECT COALESCE(SUM(count), 0) FROM daily_logs")
    total_hits = int((cur.fetchone() or [1])[0] or 1)

    cur.execute("SELECT COALESCE(SUM(l.count), 0) FROM daily_logs l JOIN paths p ON l.path_id = p.id WHERE p.value LIKE '%.md%'")
    md_reads = int((cur.fetchone() or [0])[0] or 0)
    md_pct = round((md_reads / max(1, total_hits)) * 100.0, 2)
    md_horizon_pct = min(100.0, round((md_pct / 20.0) * 100.0, 1))

    cur.execute("SELECT COALESCE(SUM(l.count), 0) FROM daily_logs l JOIN paths p ON l.path_id = p.id WHERE p.value LIKE '%js_confirm.gif%'")
    js_hits = int((cur.fetchone() or [0])[0] or 0)
    bimodal_total = max(1, js_hits + md_reads)
    js_share = round((js_hits / bimodal_total) * 100.0, 1)
    md_share = round(100.0 - js_share, 1)

    sig_clients, sig_events = db.get_rfc9421_summary()
    cneg_rows = db.get_content_neg_agents(limit=30)
    spoof_md = sum(int(c) for u, c in cneg_rows if str(u).startswith("[SPOOFED-IP]"))
    total_cneg = max(1, sum(int(c) for _, c in cneg_rows))
    spoof_pct = round((spoof_md / total_cneg) * 100.0, 1)
    auth_pct = round(100.0 - spoof_pct, 1)

    tbl = Table(expand=True, show_header=True, header_style="bold cyan", border_style="bright_blue")
    if mode == "bimodal":
        title_str = "⚡ DOM HYDRATION TRAPDOOR vs. 🧠 RFC 9110 MARKDOWN NEGOTIATORS"
        tbl.add_column("Sensor / Agent", style="bold white", ratio=3)
        tbl.add_column("Hits", justify="right", style="bold yellow", width=8)
        tbl.add_column("Share Bar (Horizontal Proportion)", ratio=4)
        js_rows = db.get_js_executors(limit=6)
        max_js = max([int(c) for _, c in js_rows] or [1])
        for ua, cnt in js_rows:
            badge = "🛑" if "[SPOOFED-IP]" in ua else ("🔷" if "[VERIFIED]" in ua else "🟠")
            short_ua = ua.replace("[SPOOFED-IP] ", "").replace("[VERIFIED] ", "").replace("Mozilla/5.0 ", "")[:34]
            tbl.add_row(f"⚡ {badge} {short_ua}", f"{cnt:,}", _bar((cnt / max_js) * 100.0, 28, "yellow"))
        tbl.add_section()
        max_md = max([int(c) for _, c in cneg_rows[:6]] or [1])
        for ua, cnt in cneg_rows[:6]:
            badge = "🛑" if "[SPOOFED-IP]" in ua else ("🔷" if "[VERIFIED]" in ua else "🟠")
            short_ua = ua.replace("[SPOOFED-IP] ", "").replace("[VERIFIED] ", "").replace("Mozilla/5.0 ", "")[:34]
            color = "red" if badge == "🛑" else ("deep_sky_blue1" if badge == "🔷" else "cyan")
            tbl.add_row(f"🧠 {badge} {short_ua}", f"{cnt:,}", _bar((cnt / max_md) * 100.0, 28, color))
    elif mode == "rfc9421":
        title_str = "🔷 RFC 9421 WEB BOT AUTH SIGNERS & 🛑 UNMASKED IP CLOAKERS"
        tbl.add_column("Operator / Signature-Agent", style="bold white", ratio=4)
        tbl.add_column("Events", justify="right", style="bold deep_sky_blue1", width=8)
        tbl.add_column("Verification Telemetry", ratio=4)
        cur.execute(
            """
            SELECT s.value, ua.value, SUM(r.count) as total
            FROM rfc9421_logs r
            JOIN sig_agents s ON r.sig_id = s.id
            JOIN user_agents ua ON r.ua_id = ua.id
            GROUP BY s.id, ua.id
            ORDER BY total DESC
            LIMIT 8
            """
        )
        sig_rows = cur.fetchall()
        max_sig = max([int(r[2]) for r in sig_rows] or [1])
        for sig_val, ua_val, cnt in sig_rows:
            short_ua = ua_val.replace("Mozilla/5.0 ", "")[:26]
            tbl.add_row(f"🔷 {sig_val} ({short_ua})", f"{cnt:,}", _bar((cnt / max_sig) * 100.0, 28, "deep_sky_blue1"))
        tbl.add_section()
        tbl.add_row(
            "🛑 Unmasked Googlebot IP Cloakers",
            f"{spoof_md:,}",
            f"{_bar(spoof_pct, 28, 'red')} [bold red]{spoof_pct}%[/bold red]",
        )
        tbl.add_row(
            "🟠 Authentic Self-Asserted Markdown Bots",
            f"{total_cneg - spoof_md:,}",
            f"{_bar(auth_pct, 28, 'dark_orange')} [bold orange1]{auth_pct}%[/bold orange1]",
        )
    else:
        title_str = "🔭 WEB BOT AI OBSERVATORY — LIVE SQLITE WIRE TELEMETRY (MIKELEV.IN)"
        tbl.add_column("Web AI Standard Metric", style="bold white", ratio=3)
        tbl.add_column("Reading", justify="right", style="bold cyan", width=18)
        tbl.add_column("Horizontal Gauge", ratio=4)
        tbl.add_row(
            "1. RFC 9110 Markdown Horizon (0% → 20%)",
            f"{md_pct}% ({md_reads:,} .md)",
            f"{_bar(md_horizon_pct, 28, 'cyan')} [cyan]{md_horizon_pct}% of 20%[/cyan]",
        )
        tbl.add_row(
            "2a. ⚡ DOM Hydration Trapdoor (JS Execs)",
            f"{js_hits:,} ({js_share}%)",
            _bar(js_share, 28, "yellow"),
        )
        tbl.add_row(
            "2b. 🧠 RFC 9110 Raw Markdown Delivery",
            f"{md_reads:,} ({md_share}%)",
            _bar(md_share, 28, "cyan"),
        )
        tbl.add_row(
            "3. 🔷 RFC 9421 Web Bot Auth (Signed)",
            f"{sig_events:,} hits · {sig_clients} clients",
            _bar(min(100.0, sig_clients * 12.0), 28, "deep_sky_blue1"),
        )
        tbl.add_row(
            "4a. 🛑 Unmasked Cloakers (Spoofed IP)",
            f"{spoof_md:,} ({spoof_pct}%)",
            _bar(spoof_pct, 28, "red"),
        )
        tbl.add_row(
            "4b. 🟠 Authentic Self-Asserted Bot UAs",
            f"{total_cneg - spoof_md:,} ({auth_pct}%)",
            _bar(auth_pct, 28, "dark_orange"),
        )

    footer = Text.from_markup(
        "\n[bold green]💸 Spending $10,000/mo+ on HTML pre-render?[/bold green] "
        "RFC 9110, RFC 9111 & RFC 9421 save that spend -> [bold cyan]https://mikelev.in/#consulting[/bold cyan]"
    )
    console.clear()
    console.print(Panel(tbl, title=f"[bold bright_cyan]{title_str}[/bold bright_cyan]", subtitle=footer, padding=(1, 2)))
    time.sleep(max(1.0, float(hold_seconds)))

if __name__ == "__main__":
    import sys as _sys
    _mode = _sys.argv[1] if len(_sys.argv) > 1 else "overview"
    render_station_report(mode=_mode, hold_seconds=65.0)
[[[REPLACE]]]

Verify the AFTER receipts for:
1. The Reverse DNS (PTR) audit of `Googlebot-Image/1.0` and `Googlebot` IPs on `js_confirm.gif` + all 6 RFC 9421 `sig_agents` in `honeybot.db`.
2. `dns_audit` and `verify_fcrdns_ip()` in `db.py`, full `Googlebot-Image` token matching and `Raw Signature` highlighting in `logs.py`.
3. The alternating Rich `WINDOW` report / `PATRONUS` ASCII art station-break rotation in `forest.py` and non-blocking voice-queued `WINDOW` support in `stream.py`.

If all receipts are green, execute the Notary Beat dismount for this article!

# # Context 1
# context.txt
# /home/mike/repos/trimnoir/_posts/2026-10-20-three-shape-bot-badges-rfc9421-svg-donut-seam.md  # [Idx: 1 | Order: 5 | Tokens: 56,679 | Bytes: 189,640]
# # foo_files.py
# prompt_foo.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/page.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/queries/trapdoor_ips.sql
# ! echo "--- WHO HYDRATES THE DOM ---" && cat remotes/honeybot/queries/trapdoor_ips.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# remotes/honeybot/queries/content_neg_agents.sql
# ! echo "--- WHO NEGOTIATES MARKDOWN ---" && cat remotes/honeybot/queries/content_neg_agents.sql | ssh honeybot 'sqlite3 -header -column ~/www/mikelev.in/honeybot.db'
# ~/repos/nixos/blogs.nix
# ~/.config/pipulate/blogs.json
# scripts/articles/publishizer.py
# scripts/articles/common.py
# scripts/articles/articleizer.py
# scripts/articles/build_knowledge_graph.py
# scripts/articles/editing_prompt.txt
# scripts/articles/sanitizer.py
# scripts/articles/contextualizer.py
# scripts/articles/generate_llms_txt.py
# scripts/articles/generate_post_nav.py
# nixops.sh
# remotes/honeybot/hooks/post-receive
# remotes/honeybot/nixos/configuration.nix
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/card.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/test_forest.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/content_loader.py
# remotes/honeybot/scripts/db.py

# # Context 2
# ! echo "--- 1. HONEYBOT ACCESS.LOG 01:39:47 EXASEARCHBOT & SIG HEADERS ---" && ssh honeybot 'grep -E "01:39:47.*ExaSearchBot" /var/log/nginx/access.log | tail -n 5; echo "--- NON-EMPTY SIG HEADERS ---"; grep "Sig:\"" /var/log/nginx/access.log | grep -v "Sig:\"-\"" | tail -n 10' && echo "--- 2. CURRENT OBSERVATORY JSON & METRICS ---" && cat /home/mike/repos/trimnoir/_data/observatory.json && rg -n "HONEYBOT_METRICS|md_horizon_pct|kpi-horizon" prompt_foo.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/_layouts/default.html /home/mike/repos/trimnoir/index.md || true
# prompt_foo.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/_layouts/default.html
# /home/mike/repos/trimnoir/_layouts/post.html
# /home/mike/repos/trimnoir/index.md
# remotes/honeybot/scripts/logs.py
# apply.py

# # Content 3
# ! echo "--- 1. CONSULTING CTA & LOGS.PY VERIFIED/REFERRAL HOOKS ---" && rg -n "10,000|verified_tokens|clean_sig_agent|src=youtube" /home/mike/repos/trimnoir/index.md /home/mike/repos/trimnoir/_layouts/post.html remotes/honeybot/scripts/logs.py || true && echo "--- 2. LOCAL PIPER TTS VOICE MODELS ON PRIME ---" && which piper 2>/dev/null || true && ls -la ~/.local/share/piper_voices/ piper_models/ 2>/dev/null || true && echo "--- 3. FOREST.PY & COMMON.PY NTFY HOOKS ---" && rg -n "STATION_SEGMENTS|send_ntfy_notification|_play_local_ding" remotes/honeybot/scripts/forest.py scripts/articles/common.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/db.py
# scripts/articles/common.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/index.md
# /home/mike/repos/trimnoir/_layouts/post.html
# apply.py

# # Context 4
# ! echo "--- 1. DB.PY & PUBLISHIZER RFC 9421 LEDGER ---" && rg -n "rfc9421_logs|sig_agents|rfc9421_signed_events|Web Bot AI Observatory" remotes/honeybot/scripts/db.py remotes/honeybot/scripts/logs.py scripts/articles/publishizer.py /home/mike/repos/trimnoir/index.md || true && echo "--- 2. FOREST 5-PART STRUCTURE ---" && rg -n "_make_station_bead|CURRENT_ARTICLE|10,000" remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/stream.py || true && echo "--- 3. COMMON.PY SPOKEN PIPER TTS TABLE ---" && rg -n "SPOKEN_ALERT_TABLE|_speak_local_alert" scripts/articles/common.py || true
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/stream.py
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/db.py
# scripts/articles/common.py
# scripts/articles/publishizer.py
# /home/mike/repos/trimnoir/index.md
# /home/mike/repos/trimnoir/_layouts/post.html
# apply.py

# Context 5
! echo "--- 1. GOOGLEBOT-IMAGE & GOOGLEBOT IPS + REVERSE DNS ON HONEYBOT ---" && ssh honeybot 'python3 -c "
import sqlite3, socket
conn = sqlite3.connect(\"/home/mike/www/mikelev.in/honeybot.db\")
cur = conn.cursor()
cur.execute(\"\"\"
SELECT i.value, ua.value, SUM(l.count) as hits
FROM daily_logs l
JOIN ips i ON l.ip_id = i.id
JOIN user_agents ua ON l.ua_id = ua.id
JOIN paths p ON l.path_id = p.id
WHERE p.value LIKE \"%js_confirm.gif%\" AND ua.value LIKE \"%Googlebot%\"
GROUP BY i.value, ua.value ORDER BY hits DESC LIMIT 10
\"\"\")
for ip, ua, hits in cur.fetchall():
    try:
        ptr = socket.gethostbyaddr(ip)[0]
    except Exception as e:
        ptr = f\"NO_PTR({e})\"
    print(f\"{hits:4d} | {ip:15s} | PTR={ptr:42s} | {ua}\")
print(\"--- RFC 9421 SIG_AGENTS IN HONEYBOT.DB ---\")
cur.execute(\"SELECT s.value, ua.value, SUM(r.count) FROM rfc9421_logs r JOIN sig_agents s ON r.sig_id=s.id JOIN user_agents ua ON r.ua_id=ua.id GROUP BY s.id, ua.id ORDER BY SUM(r.count) DESC\")
for row in cur.fetchall():
    print(row)
"' && echo "--- 2. LOCAL DNS_AUDIT & RICH REPORT HOOKS ---" && rg -n "dns_audit|render_station_report|conjure_window_overlay" remotes/honeybot/scripts/db.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/logs.py || true
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/db.py
apply.py

PUBLIC SIDE OF THE HOUSE (THE WALL)          PRIVATE SIDE OF THE HOUSE (THE STUDIO)
┌────────────────────────────────────┐       ┌────────────────────────────────────────┐
│ NixOS + Nginx Static File Origin   │       │ Honeybot Telemetry & Broadcast Engine  │
│ • Static .html & .md text files    │       │ • Tails access.log -> honeybot.db      │
│ • RFC 9110 Accept: text/markdown   │──────▶│ • Textual HUD + Rich Bar Reports       │
│ • Zero moving parts / DoD-hardened │One-Way│ • Piper TTS Player Piano (No LLM)      │
│ • No DB port, no app server, no AI │ Pipe  │ • Streams pixels out via OBS -> RTMP   │
└────────────────────────────────────┘       └────────────────────────────────────────┘
         ▲                                                      │
         │ Hostile probes hit solid granite                     ▼
   Public Internet                                    YouTube Live (24/7)

Target: apply.py
[[[SEARCH]]]
def main():
    # THE TARGET IS ROOT-RELATIVE (convicted 2026-09-28, deed 1641: `app` typed
[[[DIVIDER]]]
def _warn_multiline_context_probes(ctx_path: str = 'context.txt') -> None:
    """Warn when context.txt contains a '! ' probe split across multiple lines.
    prompt_foo.py parses context.txt line-by-line; a multiline '! ' block leaves
    an unclosed quote on line 1 (bash exit 2) and orphan lines below it."""
    if not os.path.isfile(ctx_path):
        return
    try:
        with open(ctx_path, 'r', encoding='utf-8') as f:
            lines = f.read().splitlines()
        for idx, raw in enumerate(lines, start=1):
            s = raw.strip()
            if not s or s.startswith('#'):
                continue
            if s.startswith('! ') and (s.count("'") % 2 != 0 or s.count('"') % 2 != 0 or s.endswith('\\')):
                print(f"⚠ Warning: {ctx_path}:{idx} has an unclosed quote or backslash in a '! ' probe.")
                print("   prompt_foo.py executes '! ' lines one line at a time; collapse probes to a single line.")
    except OSError:
        pass

def main():
    # THE TARGET IS ROOT-RELATIVE (convicted 2026-09-28, deed 1641: `app` typed
[[[REPLACE]]]

Target: apply.py
[[[SEARCH]]]
    payload = sys.stdin.read()
    success = apply_search_replace_patch(payload)
    sys.exit(0 if success else 1)
[[[DIVIDER]]]
    payload = sys.stdin.read()
    success = apply_search_replace_patch(payload)
    _warn_multiline_context_probes()
    sys.exit(0 if success else 1)
[[[REPLACE]]]

Target: remotes/honeybot/scripts/db.py
[[[SEARCH]]]
    def get_rfc9421_summary(self):
        """Return (distinct_sig_agents, total_signed_events) from rfc9421_logs."""
        conn = self.get_conn()
        cur = conn.cursor()
        cur.execute("SELECT COUNT(DISTINCT sig_id), COALESCE(SUM(count), 0) FROM rfc9421_logs")
        row = cur.fetchone()
        return (int(row[0] or 0), int(row[1] or 0)) if row else (0, 0)
[[[DIVIDER]]]
    def get_rfc9421_summary(self):
        """Return (distinct_sig_agents, total_signed_events) from rfc9421_logs."""
        conn = self.get_conn()
        cur = conn.cursor()
        cur.execute("SELECT COUNT(DISTINCT sig_id), COALESCE(SUM(count), 0) FROM rfc9421_logs")
        row = cur.fetchone()
        return (int(row[0] or 0), int(row[1] or 0)) if row else (0, 0)

    def get_new_user_agents_since(self, min_ua_id=0, limit=10):
        """Return (ua_id, ua_value, first_seen, total_hits) for bots discovered after min_ua_id."""
        conn = self.get_conn()
        cur = conn.cursor()
        sql = f"""
            SELECT ua.id, ua.value, ua.first_seen, COALESCE(SUM(l.count), 0) as hits
            FROM user_agents ua
            LEFT JOIN daily_logs l ON l.ua_id = ua.id
            WHERE ua.id > ?
              {self._BROWSER_FILTER}
            GROUP BY ua.id
            ORDER BY ua.id DESC
            LIMIT ?
        """
        cur.execute(sql, (int(min_ua_id), int(limit)))
        return cur.fetchall()
[[[REPLACE]]]

! echo "--- 1. LOCAL PIPER VOICE & YTPING DRY/LIVE CHECK ---" && ls -la ~/.local/share/piper_voices/ 2>/dev/null || true && .venv/bin/python scripts/articles/common.py --broadcast ping --dry-run "src=youtube test" && ssh honeybot 'sqlite3 ~/www/mikelev.in/honeybot.db "SELECT MAX(id), COUNT(*) FROM user_agents; SELECT COUNT(*), SUM(fcrdns_ok) FROM dns_audit;"'
scripts/articles/common.py
scripts/articles/publishizer.py
remotes/honeybot/scripts/db.py
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/stream.py
remotes/honeybot/scripts/logs.py
foo_files.py
apply.py

Goodbye SVG donuts; hello 0-20% RFC 9110 horizon bars, 7 RFC 9421 Web Bot Auth signers in SQLite, FCrDNS Googlebot-Image audits, and Rich station breaks. #RFC9421 #RFC9110 #TechnicalSEO
https://mikelev.in/futureproof/web-bot-ai-observatory-rfc9421-fcrdns-horizon-bars/

A dramatic widescreen graphic-novel ink-and-wash illustration set in a deep obsidian-slate (#111111) architectural chamber divided down the center by a thick granite airlock wall. On the left public-facing side, a glowing cyan horizontal horizon gauge calibrated from 0% to 20% and a Cherenkov-blue multifaceted diamond prism refract incoming beams of network light while deflecting red octagonal stop-sign probes. A single narrow armored glass pipe passes through the center of the granite wall into the private broadcast studio on the right, where glowing amber and cyan horizontal bar charts float above a classic mechanical player-piano roll and a vintage microphone radiating acoustic waves. High-contrast rim lighting in electric cyan, Cherenkov blue, and warm amber illuminates the centered focal instruments while the dark slate stonework extends to all edges. Full-bleed edge-to-edge composition filling 100% of the canvas with no borders, no frames, no matting, no margins, and no text overlays or lettering.