fetch.sh

#!/bin/bash
# fetch.sh — the nightly "reflection" wrapper. Thin by design: it owns
# only the OPERATIONAL concerns of an unattended cron run — no overlap,
# a wall-clock timeout, liveness heartbeats, an outcome event — and then
# hands the night to the agent.
#
# Unlike v1, this wrapper is NOT a security boundary. The Reflection agent
# runs with FULL tools and a REAL token (no staging tree, no
# Bash-less/token-less envelope, no wrapper-owned validation gate). It forks
# chats, reviews Memory's update log, edits skills, fixes apps, writes
# the brief to reports/<date>.html via the storage API, and commits —
# all itself, instructed by its skill
# (/data/shared/skills/reflection.md), per Möbius's "code empowers the
# agent; it does not police it." Reversibility comes from git, not from
# walls. So this file gathers a little read-only context for the agent,
# exports the few env vars its shell needs, runs the runner under a lock
# + timeout, and records how the night finished.
#
# Invoked by cron as: /data/apps/reflection/fetch.sh <app_id>
# (the app id arrives as $1, per the cron-scaffold convention).
#
# REFLECTION_DRY=1 skips the real agent run (records a dry outcome) so the
# plumbing — lock, inputs, env, heartbeat, cron_outcome — can be smoke-
# tested without spending a nightly run.
set -uo pipefail

APP_ID="${1:-}"
API_BASE_URL="${API_BASE_URL:-http://localhost:8000}"
DATA_DIR="${DATA_DIR:-/data}"
if [[ ! "$APP_ID" =~ ^[0-9]+$ ]]; then
  echo "reflection: numeric app id required as \$1" >&2
  exit 2
fi
LOG="$DATA_DIR/cron-logs/reflection.log"
LOG_ARCHIVE="$DATA_DIR/cron-logs/reflection.log.1"
LOG_FALLBACK_ARCHIVE="$DATA_DIR/cron-logs/reflection.log.rotation-fallback"
LOCK="$DATA_DIR/cron-logs/reflection.lock"
HEARTBEAT="$DATA_DIR/cron-logs/reflection.heartbeat"
TOKEN_FILE="$DATA_DIR/service-token.txt"
DATE="$(date +%F)"
INPUTS="$DATA_DIR/apps/reflection/inputs"
SCRIPT_DIR="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)"
RUNTIME_DIR="$DATA_DIR/apps/$APP_ID"
RUNNER="${REFLECTION_RUNNER:-$SCRIPT_DIR/reflection_runner.py}"
INPUT_HELPER="$SCRIPT_DIR/reflection_inputs.py"
# Wall-clock cap for the whole night. Generous (the agent does real,
# multi-phase work) but bounded so a wedged run can't hold the lock past
# the next night's schedule. Overridable for tests.
RUN_TIMEOUT="${REFLECTION_TIMEOUT:-7200}"
# Rotate only between runs, while this wrapper owns the no-overlap lock and
# before the runner opens its long-lived append descriptor. One primary archive
# plus one fixed failure fallback is enough because run metrics/activity events
# are the durable outcome record; these files are the bounded diagnostic trace.
LOG_MAX_BYTES="${REFLECTION_LOG_MAX_BYTES:-1048576}"
[[ "$LOG_MAX_BYTES" =~ ^[0-9]+$ ]] || LOG_MAX_BYTES=1048576
RUN_METRICS="$RUNTIME_DIR/reflection-run-metrics.jsonl"
RUN_CHECKPOINT="$RUNTIME_DIR/reflection-checkpoint.json"
MODEL_USAGE="$INPUTS/model-usage.json"
LATEST_EFFORT="$INPUTS/latest-effort.json"
RUN_STARTED_AT="$(date -u +%Y-%m-%dT%H:%M:%S+00:00)"
RUN_STARTED_EPOCH="$(date +%s)"
RUN_ID="${REFLECTION_RUN_ID:-${RUN_STARTED_AT}.$$}"
RUN_DISK_BEFORE="$(python3 - "$DATA_DIR" <<'PY' 2>/dev/null || echo 0
import shutil, sys
print(shutil.disk_usage(sys.argv[1]).used)
PY
)"
RUN_CPU_BEFORE="$(awk '$1 == "usage_usec" {print $2}' /sys/fs/cgroup/cpu.stat 2>/dev/null || echo 0)"

# CLI credentials the spawned claude/codex binary reads. Exported (not
# just set) so the runner and any subprocess it forks inherit them.
export CLAUDE_CONFIG_DIR="${CLAUDE_CONFIG_DIR:-$DATA_DIR/cli-auth/claude}"
export CODEX_HOME="${CODEX_HOME:-$DATA_DIR/cli-auth/codex}"
export API_BASE_URL DATA_DIR

mkdir -p "$DATA_DIR/cron-logs"
log() { echo "[$(date -Iseconds)] reflection: $*" >>"$LOG"; }

# --- no-overlap lock (flock) ------------------------------------------
# fd 9 holds the lock for the life of this process; flock -n fails fast
# if a prior night is still running (a long run that overran its window).
# Exit-code legend (recorded as the cron_outcome exit_code, so the next
# run + the Reflection app can tell a real success from a no-op):
#   0  success           3  service token missing
#   2  app id missing    5  skipped (a prior run still holds the lock)
#   124 wall-clock timeout    other  agent run error
exec 9>"$LOCK"
if ! flock -n 9; then
  log "another reflection run holds the lock; skipping this night (exit 5)"
  exit 5
fi

# Everything below mutates run state and therefore belongs behind the lock.
# An overlapping invocation must not clear the active run's inputs before it
# discovers that the lock is held.
mkdir -p "$INPUTS" "$RUNTIME_DIR"
# Older releases kept operational receipts beside catalog source. Preserve the
# last complete state once, then keep every future write in numeric app storage
# so source review shows code rather than the previous run's cursor.
for runtime_name in \
    reflection-run-metrics.jsonl reflection-checkpoint.json \
    resource-history.jsonl resource-monitor-state.json \
    resource-decisions.jsonl meta-state.md meta-learning.jsonl \
    experiments.jsonl; do
  legacy="$DATA_DIR/apps/reflection/$runtime_name"
  canonical="$RUNTIME_DIR/$runtime_name"
  if [[ -f "$legacy" && ! -e "$canonical" ]]; then
    cp -p -- "$legacy" "$canonical"
  fi
done
rm -f -- "$MODEL_USAGE"
# Optional engagement inputs must describe THIS run's predecessor. Without
# clearing them first, a night with no new answer file would inherit an older
# answer and falsely look engaged forever.
rm -f "$INPUTS/prev-report.html" "$INPUTS/prev-report-name.txt" \
  "$INPUTS/prev-question-answers.json"

# Atomic rename, never in-place truncation. At this point no Reflection runner
# can be active (fd 9 proves exclusivity), and this wrapper has not opened LOG
# persistently. The history stage below reads both archive names plus LOG, so
# rotating does not hide last night's friction from the next agent.
rotation_error="$(python3 - "$LOG" "$LOG_ARCHIVE" "$LOG_FALLBACK_ARCHIVE" "$LOG_MAX_BYTES" 2>&1 <<'PY'
import os, pathlib, sys

path = pathlib.Path(sys.argv[1])
archive = pathlib.Path(sys.argv[2])
fallback = pathlib.Path(sys.argv[3])
limit = int(sys.argv[4])
try:
    size = path.stat().st_size
except FileNotFoundError:
    raise SystemExit(0)
if size >= limit:
    try:
        os.replace(path, archive)
    except OSError as primary_error:
        # A persistently invalid primary target (observed when .1 was a
        # directory) must not leave the oversized current file growing every
        # night. Move it to one fixed fallback name instead. Replacing that
        # fallback on later failures keeps retention bounded; this still uses
        # atomic rename and runs before any long-lived writer is opened.
        try:
            os.replace(path, fallback)
        except OSError as fallback_error:
            print(
                f"primary archive failed: {primary_error}; "
                f"fallback archive failed: {fallback_error}"
            )
            raise SystemExit(1)
        print(
            f"primary archive failed: {primary_error}; "
            f"rotated to fallback {fallback.name}"
        )
        raise SystemExit(10)
PY
)"
if [[ $? -ne 0 ]]; then
  log "WARN reflection log rotation: ${rotation_error:0:300}"
fi

# --- token: export for the agent's shell (NOT a boundary) -------------
# The agent does its own privileged work (API reads, storage writes,
# notifications, git) using this token. We export it; we do NOT mediate
# the agent's use of it. A missing token means the agent can't reach the
# API, so fail loud rather than run a crippled night.
if [[ ! -r "$TOKEN_FILE" ]]; then
  log "ERROR service token unreadable ($TOKEN_FILE) — is the instance signed out? exiting"
  exit 3
fi
SERVICE_TOKEN="$(cat "$TOKEN_FILE")"
# Both names: AGENT_TOKEN is what the skill's curl examples use; the
# wrapper-era scripts read SERVICE_TOKEN. Export both so either works.
export SERVICE_TOKEN AGENT_TOKEN="$SERVICE_TOKEN"
auth=(-H "Authorization: Bearer $SERVICE_TOKEN")

log "start (app_id=$APP_ID date=$DATE dry=${REFLECTION_DRY:-0} timeout=${RUN_TIMEOUT}s)"

# --- gather read-only inputs for the agent ----------------------------
# The agent reads these from inputs/ as its starting context. It can (and
# does) gather more itself with its token. Inputs resume at the start of the
# last completed real run, so missed, broken, or quota-blocked nights remain in
# the next bundle. Best-effort inputs carry explicit source status: a failed
# gather must not masquerade as a genuine empty observation window.

# Use the start of the last completed real run rather than yesterday's date.
# Failed and dry runs do not move the checkpoint. Starting at the prior run's
# beginning deliberately overlaps events gathered during that run rather than
# risking a gap while the agent was working.
if ! SINCE="$(
  python3 "$INPUT_HELPER" resume-since "$RUN_CHECKPOINT" "$RUN_METRICS" 2>>"$LOG"
)"; then
  SINCE="1970-01-01T00:00:00Z"
  log "WARN could not read Reflection resume checkpoint; using full history"
fi
CHECKPOINT_READY=true
ACTIVITY_STATUS="$INPUTS/activity-status.json"

write_activity_status() {
  local ok="$1" error="$2" event_count="${3:-}" sha256="${4:-}"
  python3 "$INPUT_HELPER" activity-status \
    "$ACTIVITY_STATUS" "$ok" "$error" "$event_count" "$SINCE" "$sha256"
}

record_activity_status() {
  if ! write_activity_status "$@" 2>>"$LOG"; then
    # Never leave yesterday's `ok:true` sidecar beside a failed current fetch.
    # Missing status makes the digest fail closed with an explicit source error.
    rm -f -- "$ACTIVITY_STATUS"
    log "WARN could not persist activity source status"
  fi
}

# Never stream directly over the last good snapshot. curl -f rejects HTTP
# errors, its non-zero exit catches interrupted transfers, validation rejects a
# syntactically successful non-NDJSON body, and same-directory rename installs
# the new source atomically only after all three checks pass.
# Fail closed before touching the snapshot. If the process dies between the
# snapshot rename and its success sidecar, the digest sees this in-progress
# marker rather than pairing new bytes with yesterday's ok:true status.
record_activity_status false "activity fetch in progress" ""
ACTIVITY_TMP="$(mktemp "$INPUTS/.activity.jsonl.XXXXXX" 2>>"$LOG" || true)"
if [[ -z "$ACTIVITY_TMP" ]]; then
  log "WARN activity gather could not create a temporary file"
  record_activity_status false "could not create activity download temporary file" ""
elif curl -fsS --connect-timeout 10 --max-time 60 "${auth[@]}" \
    "$API_BASE_URL/api/admin/activity?since=$SINCE" \
    >"$ACTIVITY_TMP" 2>>"$LOG"; then
  if ACTIVITY_EVENT_COUNT="$(
    python3 "$INPUT_HELPER" validate-activity "$ACTIVITY_TMP" 2>>"$LOG"
  )"; then
    if mv -f -- "$ACTIVITY_TMP" "$INPUTS/activity.jsonl"; then
      ACTIVITY_TMP=""
      ACTIVITY_SHA256="$(sha256sum "$INPUTS/activity.jsonl" | awk '{print $1}')"
      record_activity_status true "" "$ACTIVITY_EVENT_COUNT" "$ACTIVITY_SHA256"
    else
      log "WARN activity gather could not atomically install its snapshot"
      record_activity_status false "could not install validated activity snapshot" ""
    fi
  else
    log "WARN activity gather returned invalid NDJSON"
    record_activity_status false "activity response was not valid NDJSON" ""
  fi
else
  activity_curl_rc="$?"
  log "WARN activity gather failed (curl rc=$activity_curl_rc); retaining prior snapshot"
  record_activity_status false "activity fetch failed (curl exit $activity_curl_rc)" ""
fi
[[ -z "${ACTIVITY_TMP:-}" ]] || rm -f -- "$ACTIVITY_TMP"
if ! python3 - "$ACTIVITY_STATUS" <<'PY' 2>>"$LOG"
import json, sys
raise SystemExit(0 if json.load(open(sys.argv[1])).get("ok") is True else 1)
PY
then
  CHECKPOINT_READY=false
fi

# chats.md — every unreviewed active and recoverable-deleted chat, oldest first,
# plus cheap
# note/message-size signals. The helper writes a structured status receipt next
# to the human-readable digest so titles can never masquerade as fetch state.
if ! python3 "$INPUT_HELPER" chats \
  "$API_BASE_URL" "$SERVICE_TOKEN" "$DATA_DIR" \
  "$INPUTS/chats.md" "$INPUTS/chats-status.json" "$SINCE" >>"$LOG" 2>&1; then
  log "WARN chat digest staging failed"
fi
if ! python3 - "$INPUTS/chats-status.json" <<'PY' 2>>"$LOG"
import json, sys
status = json.load(open(sys.argv[1]))
raise SystemExit(
    0 if status.get("active_ok") is True and status.get("deleted_complete") is True else 1
)
PY
then
  CHECKPOINT_READY=false
fi

# app-feedback.md — cross-app feedback forms written under
# shared/app-feedback/<app-slug>/. Reflection can use these as durable
# product/editorial signals without needing to know each app's numeric id.
if ! python3 - "$API_BASE_URL" "$SERVICE_TOKEN" "$SINCE" >"$INPUTS/app-feedback.md" 2>>"$LOG" <<'PY'
import datetime, json, sys, urllib.parse, urllib.request
base, token, since = sys.argv[1].rstrip("/"), sys.argv[2], sys.argv[3]
cutoff = datetime.datetime.fromisoformat(since.replace("Z", "+00:00"))
headers = {"Authorization": "Bearer "+token}

def get_json(path):
    req = urllib.request.Request(base+path, headers=headers)
    with urllib.request.urlopen(req, timeout=20) as r:
        return json.loads(r.read().decode("utf-8"))

def changed_after(entry):
    value = entry.get("modified_at")
    if not isinstance(value, str):
        return False
    try:
        changed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
        if changed.tzinfo is None:
            changed = changed.replace(tzinfo=datetime.timezone.utc)
        return changed > cutoff
    except ValueError:
        return False

def list_entries(prefix, limit=500):
    cursor = None
    seen = set()
    entries = []
    while True:
        path = "/api/storage/shared-list/" + urllib.parse.quote(prefix.strip("/"), safe="/")
        params = {"limit": str(limit)}
        if cursor:
            params["cursor"] = cursor
        path += "?" + urllib.parse.urlencode(params)
        data = get_json(path)
        entries.extend(data.get("entries", []))
        nxt = data.get("next_cursor")
        if not nxt or nxt in seen:
            break
        seen.add(nxt)
        cursor = nxt
    return entries

print("# App feedback awaiting Reflection review (oldest first)\n")
try:
    entries = []
    read_failed = False
    app_dirs = []
    for entry in list_entries("app-feedback"):
        name = entry.get("name")
        path = entry.get("path")
        if entry.get("type") == "dir" and isinstance(path, str):
            app_dirs.append(path)
        elif entry.get("type") == "dir" and isinstance(name, str):
            app_dirs.append("app-feedback/" + name)
        elif entry.get("type") == "file" and str(name or "").endswith(".json"):
            entries.append(entry)
    for app_dir in sorted(set(app_dirs)):
        for entry in list_entries(app_dir):
            if entry.get("type") == "file" and str(entry.get("name", "")).endswith(".json"):
                entries.append(entry)
    entries = sorted(
        (entry for entry in entries if changed_after(entry)),
        key=lambda entry: entry["modified_at"],
    )
    if not entries:
        print("(no app feedback)")
    for entry in entries:
        path = entry.get("path") or f"app-feedback/{entry.get('name','')}"
        try:
            item = get_json("/api/storage/shared/" + urllib.parse.quote(path, safe="/"))
            app = item.get("app") or item.get("app_id") or "app"
            signal = item.get("signal") or "note"
            date = item.get("report_date") or item.get("created_at") or ""
            text = (item.get("text") or "").replace("\n", " ").strip()
            print(f"- [{app}] {signal} {date}: {text or '(no note)'}")
        except Exception as exc:
            read_failed = True
            print(f"- {path}: could not read ({exc})")
    if read_failed:
        raise SystemExit(1)
except Exception as e:
    print(f"(could not list app feedback: {e})")
    raise SystemExit(1)
PY
then
  CHECKPOINT_READY=false
  log "WARN app-feedback staging failed"
fi

# prev-report.html — yesterday's brief, so the agent doesn't repeat
# itself. Enumerate every cursor page and fetch the newest report.
PREV="$(API_BASE_URL="$API_BASE_URL" APP_ID="$APP_ID" SERVICE_TOKEN="$SERVICE_TOKEN" python3 - <<'PY' 2>>"$LOG"
import json, os, sys, urllib.parse, urllib.request

base = os.environ["API_BASE_URL"].rstrip("/")
app_id = os.environ["APP_ID"]
token = os.environ["SERVICE_TOKEN"]
headers = {"Authorization": f"Bearer {token}"}
cursor = None
seen = set()
reports = []

try:
    for _ in range(50):
        url = f"{base}/api/storage/apps-list/{app_id}/reports/"
        if cursor:
            url += "?" + urllib.parse.urlencode({"cursor": cursor})
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=20) as r:
            data = json.loads(r.read().decode("utf-8"))
        for entry in data.get("entries", []):
            name = entry.get("name")
            if entry.get("type") == "file" and isinstance(name, str) and name.endswith(".html"):
                reports.append(name)
        nxt = data.get("next_cursor")
        if not nxt or nxt in seen:
            break
        seen.add(nxt)
        cursor = nxt
    print(sorted(reports)[-1] if reports else "")
except Exception as exc:
    print(f"could not enumerate previous reports: {exc}", file=sys.stderr)
    print("")
PY
)"
if [[ -n "$PREV" ]]; then
  printf '%s\n' "$PREV" >"$INPUTS/prev-report-name.txt"
  curl -s "${auth[@]}" "$API_BASE_URL/api/storage/apps/$APP_ID/reports/$PREV" \
    >"$INPUTS/prev-report.html" 2>>"$LOG" || true
fi

# prev-question-answers.json — every answer packet submitted since the last
# completed Reflection run, oldest first. No live agent waited; the next
# successful run receives the whole outstanding sequence.
if ! python3 "$INPUT_HELPER" question-answers \
    "$API_BASE_URL" "$SERVICE_TOKEN" "$APP_ID" "$SINCE" \
    "$INPUTS/prev-question-answers.json" >>"$LOG" 2>&1; then
  rm -f -- "$INPUTS/prev-question-answers.json"
  CHECKPOINT_READY=false
  log "WARN question-answer staging failed"
fi

# per-app-digest.json — compact per-app analytics summary the Reflection
# agent uses to triage which apps need attention tonight. Produced from
# two sources:
#   - activity.jsonl ON DISK for opens and durable app_signal events
#   - legacy signals.jsonl only for apps that wrote it in this same window
# ~2–3 KB for 12 apps vs 10–100 KB of raw log; gives the agent a
# digest-first orientation so it doesn't burn turns re-reading raw events.
# Graceful on API errors: a targeted legacy read records has_signals:false and
# an error note rather than aborting the whole step.
DIGEST_TMP="$(mktemp "$INPUTS/.per-app-digest.json.XXXXXX" 2>>"$LOG" || true)"
if [[ -n "$DIGEST_TMP" ]] && python3 "$INPUT_HELPER" app-digest \
    "$API_BASE_URL" "$SERVICE_TOKEN" "$INPUTS" "$SINCE" \
    >"$DIGEST_TMP" 2>>"$LOG"
then
  if python3 -m json.tool "$DIGEST_TMP" >/dev/null 2>>"$LOG"; then
    if mv -f -- "$DIGEST_TMP" "$INPUTS/per-app-digest.json"; then
      DIGEST_TMP=""
    else
      log "WARN could not atomically install per-app digest; retaining prior digest"
    fi
  else
    log "WARN per-app digest generation returned invalid JSON; retaining prior digest"
  fi
else
  log "WARN per-app digest generation failed; retaining prior digest"
fi
[[ -z "${DIGEST_TMP:-}" ]] || rm -f -- "$DIGEST_TMP"
if ! python3 - "$INPUTS/per-app-digest.json" "$SINCE" <<'PY' 2>>"$LOG"
import json, sys
source = json.load(open(sys.argv[1])).get("activity_source") or {}
raise SystemExit(
    0 if source.get("ok") is True and source.get("since") == sys.argv[2] else 1
)
PY
then
  CHECKPOINT_READY=false
fi

# tool-friction.json — one bounded, read-only view of repeated agent plumbing.
# Reflection uses this to choose common owning-primitives worth improving rather
# than reconstructing thousands of tool blocks or promoting one-off failures.
TOOL_FRICTION="$SCRIPT_DIR/tool_friction.py"
if [[ -r "$TOOL_FRICTION" ]]; then
  if ! python3 "$TOOL_FRICTION" \
      --db "$DATA_DIR/db/ultimate.db" \
      --since "$SINCE" \
      --output "$INPUTS/tool-friction.json" 2>>"$LOG"; then
    CHECKPOINT_READY=false
    log "WARN tool-friction evidence generation failed"
  fi
else
  CHECKPOINT_READY=false
  log "WARN tool-friction helper missing at $TOOL_FRICTION"
  rm -f -- "$INPUTS/tool-friction.json"
fi

# housekeeping.json — deterministic contribution/worktree lifecycle work runs
# BEFORE the agent and BEFORE the resource snapshot. Exact merged heads can be
# retired without spending model turns; patch-equivalent but unreferenced work
# is classified for judgment rather than treated as deletion authority.
# Anything dirty, active, actionable, standalone, or ambiguous is likewise
# handed to Reflection as a compact exception list. The wrapper's dry mode
# audits without mutating so plumbing tests never remove fixtures.
HOUSEKEEPING="$SCRIPT_DIR/housekeeping.py"
HOUSEKEEPING_OUTPUT="$INPUTS/housekeeping.json"
# A producer crash must not leave yesterday's success available under today's
# path. The manifest below will record absence if even the failure payload
# cannot be written.
rm -f -- "$HOUSEKEEPING_OUTPUT"
if [[ -r "$HOUSEKEEPING" ]]; then
  housekeeping_args=(
    --data-dir "$DATA_DIR"
    --api-base-url "$API_BASE_URL"
    --token-file "$TOKEN_FILE"
    --output "$HOUSEKEEPING_OUTPUT"
  )
  if [[ "${REFLECTION_DRY:-0}" != "1" ]]; then
    housekeeping_args+=(--apply)
  fi
  if ! python3 "$HOUSEKEEPING" "${housekeeping_args[@]}" 2>>"$LOG"; then
    log "WARN deterministic housekeeping failed; inspect housekeeping.json"
  fi
else
  log "WARN deterministic housekeeping helper missing at $HOUSEKEEPING"
  rm -f -- "$HOUSEKEEPING_OUTPUT"
fi

# memory-health.json — a compact operational contract between the two apps.
# It exposes status, recovery/backlog counters, and graph counts only: never
# chat bodies, note contents, proposed facts, or other private Memory data.
MEMORY_HEALTH="$SCRIPT_DIR/memory_health.py"
if [[ -r "$MEMORY_HEALTH" ]]; then
  if ! python3 "$MEMORY_HEALTH" \
      --memory-root "$DATA_DIR/shared/memory" \
      --since "$SINCE" \
      --output "$INPUTS/memory-health.json" 2>>"$LOG"; then
    CHECKPOINT_READY=false
    log "WARN Memory health handoff failed"
  fi
else
  CHECKPOINT_READY=false
  log "WARN Memory health helper missing at $MEMORY_HEALTH"
fi

# Memory owns this profile. Reflection receives a bounded read-only snapshot
# for relevance ranking and never writes it back.
PROFILE_HANDOFF="$SCRIPT_DIR/personalization_profile.py"
if [[ -r "$PROFILE_HANDOFF" ]]; then
  if ! python3 "$PROFILE_HANDOFF" --api-base-url "$API_BASE_URL" \
      --token-file "$TOKEN_FILE" --output "$INPUTS/personalization-profile.json" 2>>"$LOG"; then
    log "WARN personalization profile handoff failed"
  fi
else
  log "WARN personalization profile helper missing at $PROFILE_HANDOFF"
fi

# resource-snapshot.json — a cheap daily filesystem/cgroup pulse plus an
# adaptive deep /data inventory. The helper remembers its last complete deep
# scan and only walks the tree weekly, under pressure, or after unusual growth.
# This gives Reflection trend evidence without paying for the same broad `du`
# commands every night. History and decisions are bounded durable logs.
RESOURCE_MONITOR="$SCRIPT_DIR/resource_monitor.py"
RESOURCE_HISTORY="$RUNTIME_DIR/resource-history.jsonl"
RESOURCE_STATE="$RUNTIME_DIR/resource-monitor-state.json"
RESOURCE_LEDGER="$RUNTIME_DIR/resource-decisions.jsonl"
META_STATE="$RUNTIME_DIR/meta-state.md"
META_LOG="$RUNTIME_DIR/meta-learning.jsonl"
EXPERIMENT_LEDGER="$RUNTIME_DIR/experiments.jsonl"
EXPERIMENT_STATUS="$SCRIPT_DIR/experiment_status.py"
if [[ -r "$RESOURCE_MONITOR" ]]; then
  if ! python3 "$RESOURCE_MONITOR" snapshot \
      --data-dir "$DATA_DIR" \
      --runtime-root / \
      --memory-api "$API_BASE_URL" \
      --token-file "$TOKEN_FILE" \
      --output "$INPUTS/resource-snapshot.json" \
      --history "$RESOURCE_HISTORY" \
      --state "$RESOURCE_STATE" 2>>"$LOG"; then
    log "WARN resource monitor failed; resource snapshot may be stale"
  fi
else
  log "WARN resource monitor missing at $RESOURCE_MONITOR"
fi
if [[ -r "$RESOURCE_HISTORY" ]]; then
  tail -n 30 "$RESOURCE_HISTORY" >"$INPUTS/resource-history.jsonl" 2>>"$LOG" || true
else
  : >"$INPUTS/resource-history.jsonl"
fi
if [[ -r "$RESOURCE_LEDGER" ]]; then
  tail -n 100 "$RESOURCE_LEDGER" >"$INPUTS/resource-decisions.jsonl" 2>>"$LOG" || true
else
  : >"$INPUTS/resource-decisions.jsonl"
fi

# A compact, agent-owned operating model. Unlike the raw run log, this file is
# deliberately rewritten as understanding evolves: it keeps only the current
# useful model of the partner, system, watchlist, and Reflection's own approach.
# The append-only learning tail preserves why that model changed.
if [[ ! -f "$META_STATE" ]]; then
  python3 - "$META_STATE" <<'PY' 2>>"$LOG" || true
import os, pathlib, sys
path = pathlib.Path(sys.argv[1])
path.parent.mkdir(parents=True, exist_ok=True)
text = """# Reflection operating model
<!-- first-run seed: never populated yet — cold-start scaffold, not a reset -->

This is the cold-start seed for Reflection's operating model. It has never been
populated yet: a fresh scaffold written on install, NOT a blanked or reset file.
The "No stable pattern recorded yet." lines below are placeholders — do not read
them as lost prior state or open a watch about a reset. Delete this note and the
sentinel comment above once you record the first real observation. Keep this
concise and evidence-based, and rewrite it as the model changes.

## Partner and working patterns
- No stable pattern recorded yet.

## System and workflow
- No stable pattern recorded yet.

## Near-term horizon
- Review recent work and open loops before predicting what may help next.

## Watchlist and cadence
- Add only useful watches, each with evidence, last checked, and next review.

## Reflection approach
- Prefer a few high-leverage, verifiable improvements over completing a checklist.
"""
tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
tmp.write_text(text, encoding="utf-8")
os.replace(tmp, path)
PY
fi
cp "$META_STATE" "$INPUTS/meta-state.md" 2>>"$LOG" || true
python3 - "$META_STATE" >"$INPUTS/meta-state-status.json" 2>>"$LOG" <<'PY' || true
import datetime as dt, hashlib, json, pathlib, sys

path = pathlib.Path(sys.argv[1])
try:
    raw = path.read_bytes()
    stat = path.stat()
except OSError:
    raw = b""
    stat = None
now = dt.datetime.now(dt.timezone.utc)
mtime = dt.datetime.fromtimestamp(stat.st_mtime, dt.timezone.utc) if stat else None
text = raw.decode("utf-8", errors="replace")
print(json.dumps({
    "version": 1,
    "canonical_live_path": str(path),
    "exists": stat is not None,
    "bytes": len(raw),
    "sha256": hashlib.sha256(raw).hexdigest() if stat else None,
    "modified_at": mtime.isoformat() if mtime else None,
    "age_hours": round((now - mtime).total_seconds() / 3600, 2) if mtime else None,
    "first_run_seed": "first-run seed" in text,
    "empty": not text.strip(),
}, indent=2, sort_keys=True))
PY
if [[ -r "$META_LOG" ]]; then
  # Keep the durable explanation log useful but finite. Reflection appends
  # JSON objects; the wrapper validates them, keeps the newest 200 records,
  # and enforces a 1 MiB ceiling before staging the recent tail.
  python3 - "$META_LOG" <<'PY' 2>>"$LOG" || true
import json, os, pathlib, sys

path = pathlib.Path(sys.argv[1])
try:
    raw_lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
except OSError:
    raise SystemExit(0)

rows = []
for raw in raw_lines:
    try:
        row = json.loads(raw)
    except (TypeError, ValueError):
        continue
    if isinstance(row, dict):
        rows.append(json.dumps(row, ensure_ascii=False, separators=(",", ":")))

rows = rows[-200:]
while rows and len(("\n".join(rows) + "\n").encode("utf-8")) > 1024 * 1024:
    rows.pop(0)

text = "\n".join(rows) + ("\n" if rows else "")
tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
tmp.write_text(text, encoding="utf-8")
os.replace(tmp, path)
PY
  tail -n 100 "$META_LOG" >"$INPUTS/meta-learning.jsonl" 2>>"$LOG" || true
else
  : >"$INPUTS/meta-learning.jsonl"
fi

# Experiments are distinct from durable lessons. The append-only ledger records
# lifecycle events under a stable experiment_id; this deterministic view tells
# the agent what is active or time-due without deciding whether an experiment
# was good, whether a trigger fired, or what outcome means.
if [[ -r "$EXPERIMENT_LEDGER" ]]; then
  tail -n 200 "$EXPERIMENT_LEDGER" >"$INPUTS/experiments.jsonl" 2>>"$LOG" || true
else
  : >"$INPUTS/experiments.jsonl"
fi
if [[ -r "$EXPERIMENT_STATUS" ]]; then
  if ! python3 "$EXPERIMENT_STATUS" \
      --ledger "$EXPERIMENT_LEDGER" \
      --output "$INPUTS/experiment-status.json" 2>>"$LOG"; then
    log "WARN experiment status handoff failed"
  fi
else
  log "WARN experiment status helper missing at $EXPERIMENT_STATUS"
fi

# reflection-run-history.txt — bounded self-observation for the next agent.
# Metrics answer "did the last change make the run cheaper?"; the normalized
# semantic tail carries friction without letting hundreds of SDK output/text
# deltas crowd it out; git history prevents a later run from re-adding an
# experiment that an earlier run deliberately removed.
python3 - "$RUN_METRICS" "$LOG_ARCHIVE" "$LOG_FALLBACK_ARCHIVE" "$LOG" "$DATA_DIR" >"$INPUTS/reflection-run-history.txt" 2>>"$LOG" <<'PY' || true
import json, pathlib, subprocess, sys

(
    metrics_path, archive_path, fallback_archive_path, log_path, data_dir,
) = map(pathlib.Path, sys.argv[1:])
print("# Reflection run history (bounded; newest last)\n")
print("## Run metrics")
try:
    rows = metrics_path.read_text(encoding="utf-8", errors="replace").splitlines()[-14:]
except OSError:
    rows = []
print("\n".join(rows) if rows else "(no prior metrics)")
print("\n## Recent reflection log tail (normalized)")

def tail_lines(path, byte_limit=2 * 1024 * 1024):
    try:
        with path.open("rb") as f:
            f.seek(0, 2)
            size = f.tell()
            start = max(0, size - byte_limit)
            f.seek(start)
            raw = f.read()
    except OSError:
        return []
    if start:
        _, _, raw = raw.partition(b"\n")
    return raw.decode("utf-8", errors="replace").splitlines()

def cap_line(line, limit=800):
    line = line.rstrip()
    if len(line) <= limit:
        return line
    return f"{line[:limit]}... [{len(line)} chars]"

def normalize(lines):
    rows = []
    text_parts = []
    output_chunks = 0

    def flush_text():
        nonlocal text_parts
        if text_parts:
            joined = " ".join(part.strip() for part in text_parts if part.strip())
            if joined:
                rows.append(cap_line(f"  > {joined}"))
        text_parts = []

    def flush_output():
        nonlocal output_chunks
        if output_chunks:
            rows.append(
                "  · codex legacy tool_output stream: "
                f"{output_chunks} chunks omitted from run history; "
                "inspect reflection.log(.1) for the raw legacy trace"
            )
        output_chunks = 0

    for line in lines:
        # Legacy _LogBroadcast serialized each streamed output event. Those
        # lines were commonly sliced mid-JSON at 500 chars, so parsing them is
        # neither reliable nor useful. Count contiguous bursts instead.
        if line.startswith('  · codex {"type": "tool_output"'):
            flush_text()
            output_chunks += 1
            continue
        flush_output()
        if line.startswith("  > ") and not line.startswith("  > codex message "):
            text_parts.append(line[4:])
            continue
        flush_text()
        rows.append(cap_line(line))
    flush_output()
    flush_text()
    return rows

archives = sorted(
    (archive_path, fallback_archive_path),
    key=lambda path: path.stat().st_mtime_ns if path.is_file() else -1,
)
raw_lines = []
for path in archives:
    raw_lines.extend(tail_lines(path))
raw_lines.extend(tail_lines(log_path))
lines = normalize(raw_lines)[-100:]
while lines and len(("\n".join(lines) + "\n").encode("utf-8")) > 32 * 1024:
    lines.pop(0)
print("\n".join(lines) if lines else "(no prior log)")
print("\n## Recent edits to reflection.md")
try:
    result = subprocess.run(
        ["git", "-C", str(data_dir), "log", "--oneline", "-10", "--",
         "shared/skills/reflection.md"],
        text=True, capture_output=True, timeout=10, check=False,
    )
    print(result.stdout.strip() or "(no recorded edits)")
except Exception as exc:
    print(f"(could not read skill history: {exc})")
PY

# Record the app id where the runner's goal message and the agent can
# find it (the agent writes reports to apps/<app_id>/reports/).
printf '%s\n' "$APP_ID" >"$INPUTS/app_id"
EFFORT_SUMMARY="$SCRIPT_DIR/effort_summary.py"
if [[ -r "$EFFORT_SUMMARY" ]]; then
  if ! python3 "$EFFORT_SUMMARY" --metrics "$RUN_METRICS" \
      --output "$INPUTS/effort-summary.json" 2>>"$LOG"; then
    log "WARN rolling effort summary failed"
  fi
else
  log "WARN effort summary helper missing at $EFFORT_SUMMARY"
fi
LEARNING_LOOP="$SCRIPT_DIR/learning_loop.py"
if [[ -r "$LEARNING_LOOP" ]]; then
  if ! python3 "$LEARNING_LOOP" --inputs "$INPUTS" \
      --output "$INPUTS/learning-loop.json" 2>>"$LOG"; then
    log "WARN consolidated learning-loop orientation failed"
  fi
else
  log "WARN learning-loop helper missing at $LEARNING_LOOP"
fi
if ! python3 "$INPUT_HELPER" manifest \
    "$INPUTS" "$RUN_ID" "$RUN_STARTED_AT" 2>>"$LOG"; then
  rm -f -- "$INPUTS/input-manifest.json"
  log "WARN could not write the Reflection input manifest"
fi
log "gathered one manifested input bundle (meta model, activity, chats, feedback, app digest, tool friction, housekeeping, resource evidence) into $INPUTS/"

# --- heartbeat: prove liveness while the long run is in flight --------
# A background loop touches the heartbeat file every 60s. A monitor (or a
# morning glance) can `stat` it to tell "still reflection" from "wedged".
# Killed in the cleanup trap below.
#
# fd 9 (the flock handle) is CLOSED in the child (`9>&-`) so the lock is
# held ONLY by the main process. Without this, the backgrounded child
# inherits fd 9 and keeps the lock alive past the parent's exit until the
# child is reaped — so the NEXT night's run would spuriously see "another
# run holds the lock" and skip. The cleanup trap kills the child and
# waits for it so the lock is fully released by the time we exit.
heartbeat_loop() {
  local sleep_pid=""
  # A backgrounded shell function gets its own PID, but a plain `sleep 60`
  # inside it is a separate child. Killing only the function used to orphan
  # that sleep until its timeout (and kept captured stdout pipes open in tests).
  # Retire the active child before the heartbeat process exits.
  trap '[[ -z "$sleep_pid" ]] || kill "$sleep_pid" 2>/dev/null || true; exit 0' TERM INT
  while true; do
    date -Iseconds >"$HEARTBEAT" 2>/dev/null || true
    sleep 60 &
    sleep_pid=$!
    wait "$sleep_pid" 2>/dev/null || true
    sleep_pid=""
  done
}
heartbeat_loop 9>&- &
HEARTBEAT_PID=$!
cleanup() {
  if [[ -n "${HEARTBEAT_PID:-}" ]]; then
    kill "$HEARTBEAT_PID" 2>/dev/null || true
    wait "$HEARTBEAT_PID" 2>/dev/null || true
  fi
}
trap cleanup EXIT

# --- run the agent: full tools, real token, no sandbox ----------------
# The runner loads the reflection skill as the system prompt, sends the
# goal as the first user message, and drives the multi-turn loop. `timeout`
# bounds wall-clock; --signal=TERM gives the run a chance to flush before
# SIGKILL (--kill-after). The runner streams its own trace into $LOG.
RC=0
if [[ "${REFLECTION_DRY:-0}" == "1" ]]; then
  log "DRY run: skipping agent; recording dry outcome"
  RC=0
elif [[ ! -r "$RUNNER" ]]; then
  log "ERROR runner not found/readable at $RUNNER; exiting"
  RC=127
else
  timeout --signal=TERM --kill-after=60 "$RUN_TIMEOUT" \
    python3 "$RUNNER" >>"$LOG" 2>&1
  RC=$?
  if [[ "$RC" == "124" ]]; then
    log "WARN agent run hit the ${RUN_TIMEOUT}s timeout (terminated)"
  elif [[ "$RC" != "0" ]]; then
    log "WARN agent run exited non-zero (rc=$RC)"
  fi
fi

# --- deterministic morning-brief push ---------------------------------
# Delivery of the morning push is owned HERE, by the wrapper — NOT by the
# agent. The agent composes the brief and writes state.json (streak +
# one-line `last_summary` headline for the app header); the wrapper reads
# that headline and fires the push via the notifications API with the
# service token.
#
# Why the wrapper and not the agent: an agent-chosen notification tool
# proved unreliable. From 2026-06-30 the nightly agent began reaching for
# a leaked Claude Code harness `PushNotification` tool (found via
# `ToolSearch: select:PushNotification`) instead of the documented
# `curl /api/notifications/send`. That harness tool is a no-op inside
# Möbius, so no morning brief reached the partner for a week even though
# every run succeeded and every brief was written. Making the wrapper the
# sole sender — exactly as news/fetch.sh already does — removes the
# dependency on the agent picking the right tool. Best-effort: a failed
# push is logged, never fatal.
send_morning_push() {
  [[ "${REFLECTION_DRY:-0}" != "1" ]] || {
    log "morning push: skip (dry run)"
    return 0
  }
  [[ "$RC" == "0" ]] || { log "morning push: skip (rc=$RC)"; return 0; }
  local brief="$DATA_DIR/apps/$APP_ID/reports/$DATE.html"
  [[ -f "$brief" ]] || { log "morning push: skip (no brief for $DATE)"; return 0; }
  # Idempotence guard: the wrapper is the sole intended sender, but an
  # instance whose live (agent-editable) reflection skill predates
  # wrapper-owned delivery may still have the nightly agent curl
  # /api/notifications/send itself mid-run — the skill file is seeded once
  # and never overwritten by app updates, so that stale instruction
  # survives upgrades indefinitely and the partner gets the same brief
  # announced twice. Skip when an identically-titled push already went out
  # today (UTC). Best-effort and fail-open: if history can't be read, send
  # rather than risk a silent morning.
  local history_json already_sent
  history_json="$(curl -s "${auth[@]}" \
    "$API_BASE_URL/api/notifications?limit=30" 2>>"$LOG")"
  already_sent="$(HISTORY_JSON="$history_json" python3 - <<'PY' 2>>"$LOG"
import datetime, json, os
try:
    items = json.loads(os.environ.get("HISTORY_JSON", ""))
except Exception:
    items = []
today = datetime.datetime.now(datetime.timezone.utc).date().isoformat()
dup = isinstance(items, list) and any(
    isinstance(n, dict)
    and n.get("title") == "Your morning brief is ready"
    and str(n.get("sent_at", "")).startswith(today)
    for n in items
)
print("yes" if dup else "no")
PY
)"
  if [[ "$already_sent" == "yes" ]]; then
    log "morning push: skip (an identical push already went out today)"
    return 0
  fi
  # Trust the headline only if state.json was written by TODAY's run;
  # fall back to a generic line otherwise so the partner is still pinged.
  local headline
  headline="$(APP_ID="$APP_ID" DATE="$DATE" DATA_DIR="$DATA_DIR" python3 - <<'PY' 2>>"$LOG"
import json, os
try:
    s = json.load(open(f"{os.environ['DATA_DIR']}/apps/{os.environ['APP_ID']}/state.json"))
except Exception:
    s = {}
head = (s.get("last_summary") or "").strip()
print(head if str(s.get("last_run", "")).startswith(os.environ["DATE"]) else "")
PY
)"
  [[ -n "$headline" ]] || headline="Your nightly reflection is ready to read."
  local payload
  payload="$(APP_ID="$APP_ID" HEADLINE="$headline" python3 - <<'PY' 2>>"$LOG"
import json, os
app_id = os.environ["APP_ID"]
target = f"/shell/?app={app_id}"
print(json.dumps({
    "title": "Your morning brief is ready",
    "body": os.environ["HEADLINE"][:200],
    "source_type": "app",
    "source_id": app_id,
    "target": target,
    "actions": [{"action": "open_app", "title": "Read", "target": target}],
}))
PY
)"
  local code
  code="$(curl -s -o /dev/null -w '%{http_code}' -X POST "${auth[@]}" \
    -H "Content-Type: application/json" -d "$payload" \
    "$API_BASE_URL/api/notifications/send" 2>>"$LOG")"
  case "$code" in
    200|201|204) log "morning push sent (http=$code)";;
    *)           log "WARN morning push failed (http=$code)";;
  esac
}
send_morning_push

# Persist one compact row about Reflection's own footprint. Keep the log
# bounded: it is evidence for qualitative review, not an optimization target.
RUN_FINISHED_AT="$(date -u +%Y-%m-%dT%H:%M:%S+00:00)"
RUN_FINISHED_EPOCH="$(date +%s)"
RUN_DISK_AFTER="$(python3 - "$DATA_DIR" <<'PY' 2>/dev/null || echo 0
import shutil, sys
print(shutil.disk_usage(sys.argv[1]).used)
PY
)"
RUN_CPU_AFTER="$(awk '$1 == "usage_usec" {print $2}' /sys/fs/cgroup/cpu.stat 2>/dev/null || echo 0)"
python3 - "$RUN_METRICS" "$RUN_STARTED_AT" "$RUN_FINISHED_AT" \
    "$RUN_STARTED_EPOCH" "$RUN_FINISHED_EPOCH" "$RC" \
    "$RUN_DISK_BEFORE" "$RUN_DISK_AFTER" "$RUN_CPU_BEFORE" "$RUN_CPU_AFTER" \
    "$DATA_DIR/apps/$APP_ID/reports/$DATE.html" "${REFLECTION_DRY:-0}" \
    "$MODEL_USAGE" "$LATEST_EFFORT" \
    2>>"$LOG" <<'PY' || log "WARN could not persist reflection run metrics"
import json, os, pathlib, sys

(path, started_at, finished_at, started_epoch, finished_epoch, rc,
 disk_before, disk_after, cpu_before, cpu_after, report, dry,
 model_usage, latest_effort) = sys.argv[1:]
def integer(value):
    try: return int(value)
    except (TypeError, ValueError): return 0
row = {
    "started_at": started_at,
    "finished_at": finished_at,
    "duration_seconds": max(0, integer(finished_epoch) - integer(started_epoch)),
    "exit_code": integer(rc),
    "dry_run": dry == "1",
    "brief_written": (
        pathlib.Path(report).is_file()
        and pathlib.Path(report).stat().st_mtime >= integer(started_epoch)
    ),
    "disk_used_delta_bytes": integer(disk_after) - integer(disk_before),
    "cgroup_cpu_usage_usec_delta": max(0, integer(cpu_after) - integer(cpu_before)),
}
try:
    receipt = json.load(open(model_usage, encoding="utf-8"))
except (OSError, ValueError):
    receipt = None
if isinstance(receipt, dict):
    row["model_usage"] = receipt
target = pathlib.Path(path)
try:
    lines = target.read_text(encoding="utf-8", errors="replace").splitlines()
except OSError:
    lines = []
lines = [line for line in lines if line.strip()][-59:]
lines.append(json.dumps(row, ensure_ascii=False, separators=(",", ":")))
target.parent.mkdir(parents=True, exist_ok=True)
tmp = target.with_name(f".{target.name}.{os.getpid()}.tmp")
with tmp.open("w", encoding="utf-8") as handle:
    handle.write("\n".join(lines) + "\n")
    handle.flush(); os.fsync(handle.fileno())
os.replace(tmp, target)
if dry != "1":
    latest = pathlib.Path(latest_effort)
    latest_tmp = latest.with_name(f".{latest.name}.{os.getpid()}.tmp")
    latest_tmp.write_text(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n")
    os.replace(latest_tmp, latest)
PY

checkpoint_result="$(python3 "$INPUT_HELPER" complete-run \
  "$RUN_CHECKPOINT" "$RUN_STARTED_AT" \
  "$DATA_DIR/apps/$APP_ID/reports/$DATE.html" "$RC" \
  "$([[ "${REFLECTION_DRY:-0}" == "1" ]] && printf true || printf false)" \
  "$CHECKPOINT_READY" \
  2>>"$LOG" || true)"
if [[ "$RC" == "0" && "${REFLECTION_DRY:-0}" != "1" && "$checkpoint_result" != "true" ]]; then
  log "WARN successful run did not advance its checkpoint (current brief missing or stale)"
fi

log "done (rc=$RC)"
exit "$RC"