fix: wrap breaker calls in stage_loop to prevent permanent task death

A transient DB lock in breaker.record_failure() inside an except handler
killed the asyncio coroutine permanently — snapshot_cycle died Apr 18 and
never recovered. All three breaker call sites now have their own try/except.

Also includes HTML injection fix for github_feedback review_text.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

deploy/auto-deploy.sh

@@ -78,6 +78,9 @@ rsync $RSYNC_FLAGS diagnostics/ "$DIAGNOSTICS_DIR/"
 rsync $RSYNC_FLAGS agent-state/ "$AGENT_STATE_DIR/"
 [ -f research/research-session.sh ] && rsync $RSYNC_FLAGS research/research-session.sh /opt/teleo-eval/research-session.sh
 
+# Safety net: ensure all .sh files are executable after rsync
+find /opt/teleo-eval -maxdepth 3 -name '*.sh' -not -perm -u+x -exec chmod +x {} +
+
 log "Files synced"
 
 # Restart services only if Python files changed

deploy/sync-mirror.sh

@@ -191,10 +191,11 @@ print('no')
                 else
                     PR_TITLE=$(echo "$branch" | sed 's|/|: |;s/-/ /g')
                 fi
+                PAYLOAD=$(python3 -c "import sys,json; print(json.dumps({'title':sys.argv[1],'head':sys.argv[2],'base':'main'}))" "$PR_TITLE" "$branch")
                 RESULT=$(curl -sf -X POST "http://localhost:3000/api/v1/repos/teleo/teleo-codex/pulls" \
                     -H "Authorization: token $FORGEJO_TOKEN" \
                     -H "Content-Type: application/json" \
-                    -d "{\"title\":\"$PR_TITLE\",\"head\":\"$branch\",\"base\":\"main\"}" 2>/dev/null || echo "")
+                    -d "$PAYLOAD" 2>/dev/null || echo "")
                 PR_NUM=$(echo "$RESULT" | grep -o '"number":[0-9]*' | head -1 | grep -o "[0-9]*" || true)
                 if [ -n "$PR_NUM" ]; then
                     log "Auto-created PR #$PR_NUM on Forgejo for $branch"
@@ -210,7 +211,7 @@ print('no')
                                 python3 -c "import sys,json; prs=json.load(sys.stdin); print(prs[0]['number'] if prs else '')" 2>/dev/null || true)
                         fi
                     fi
-                    if [ -n "$GH_PR_NUM" ]; then
+                    if [[ "$GH_PR_NUM" =~ ^[0-9]+$ ]] && [[ "$PR_NUM" =~ ^[0-9]+$ ]]; then
                         sqlite3 "$PIPELINE_DB" "UPDATE prs SET github_pr = $GH_PR_NUM WHERE number = $PR_NUM;" 2>/dev/null && \
                             log "Linked GitHub PR #$GH_PR_NUM -> Forgejo PR #$PR_NUM" || \
                             log "WARN: Failed to link GitHub PR #$GH_PR_NUM to Forgejo PR #$PR_NUM in DB"
@@ -225,4 +226,57 @@ else
     log "No new GitHub-only branches"
 fi
 
+# Step 6: Divergence alerting
+# After all sync steps, check if GitHub and Forgejo main still differ.
+# 2 consecutive divergent cycles (4 min) triggers a one-shot Telegram alert.
+DIVERGENCE_FILE="/opt/teleo-eval/logs/.divergence-count"
+git fetch forgejo main --quiet 2>/dev/null || true
+git fetch origin main --quiet 2>/dev/null || true
+GH_MAIN_FINAL=$(git rev-parse refs/remotes/origin/main 2>/dev/null || true)
+FG_MAIN_FINAL=$(git rev-parse refs/remotes/forgejo/main 2>/dev/null || true)
+
+if [ -n "$GH_MAIN_FINAL" ] && [ -n "$FG_MAIN_FINAL" ] && [ "$GH_MAIN_FINAL" != "$FG_MAIN_FINAL" ]; then
+    PREV=$(cat "$DIVERGENCE_FILE" 2>/dev/null || echo "0")
+    if [ "$PREV" = "alerted" ]; then
+        log "DIVERGENCE: still diverged (already alerted)"
+    else
+        COUNT=$((PREV + 1))
+        echo "$COUNT" > "$DIVERGENCE_FILE"
+        log "DIVERGENCE: cycle $COUNT — GitHub=$GH_MAIN_FINAL Forgejo=$FG_MAIN_FINAL"
+        if [ "$COUNT" -ge 2 ]; then
+            BOT_TOKEN=$(cat /opt/teleo-eval/secrets/telegram-bot-token 2>/dev/null || true)
+            ADMIN_CHAT=$(cat /opt/teleo-eval/secrets/admin-chat-id 2>/dev/null || true)
+            if [ -n "$BOT_TOKEN" ] && [ -n "$ADMIN_CHAT" ]; then
+                ALERT_MSG=$(python3 -c "
+import json, sys
+msg = '⚠️ Mirror divergence detected\\n\\n'
+msg += f'GitHub main: {sys.argv[1][:8]}\\n'
+msg += f'Forgejo main: {sys.argv[2][:8]}\\n'
+msg += f'Diverged for {sys.argv[3]} consecutive cycles ({int(sys.argv[3])*2} min)\\n\\n'
+msg += 'Check sync-mirror.sh logs: /opt/teleo-eval/logs/sync.log'
+print(json.dumps({'chat_id': sys.argv[4], 'text': msg, 'parse_mode': 'HTML'}))
+" "$GH_MAIN_FINAL" "$FG_MAIN_FINAL" "$COUNT" "$ADMIN_CHAT")
+                if curl -sf -X POST "https://api.telegram.org/bot${BOT_TOKEN}/sendMessage" \
+                    -H "Content-Type: application/json" \
+                    -d "$ALERT_MSG" >> "$LOG" 2>&1; then
+                    log "DIVERGENCE: alert sent to admin"
+                    echo "alerted" > "$DIVERGENCE_FILE"
+                else
+                    log "WARN: Failed to send divergence alert (will retry next cycle)"
+                fi
+            else
+                log "WARN: Cannot send divergence alert — missing bot token or admin chat ID"
+            fi
+        fi
+    fi
+else
+    if [ -f "$DIVERGENCE_FILE" ]; then
+        PREV=$(cat "$DIVERGENCE_FILE" 2>/dev/null || echo "0")
+        if [ "$PREV" != "0" ]; then
+            log "DIVERGENCE: resolved — repos back in sync"
+        fi
+        rm -f "$DIVERGENCE_FILE"
+    fi
+fi
+
 log "Sync complete"

lib/db.py

@@ -9,7 +9,7 @@ from . import config
 
 logger = logging.getLogger("pipeline.db")
 
-SCHEMA_VERSION = 21
+SCHEMA_VERSION = 22
 
 SCHEMA_SQL = """
 CREATE TABLE IF NOT EXISTS schema_version (
@@ -71,6 +71,7 @@ CREATE TABLE IF NOT EXISTS prs (
     cost_usd REAL DEFAULT 0,
     auto_merge INTEGER DEFAULT 0,
     github_pr INTEGER,
+    source_channel TEXT,
     created_at TEXT DEFAULT (datetime('now')),
     merged_at TEXT
 );
@@ -196,6 +197,7 @@ def transaction(conn: sqlite3.Connection):
 # Branch prefix → (agent, commit_type) mapping.
 # Single source of truth — used by merge.py at INSERT time and migration v7 backfill.
 # Unknown prefixes → ('unknown', 'unknown') + warning log.
+# Keep in sync with _CHANNEL_MAP below.
 BRANCH_PREFIX_MAP = {
     "extract": ("pipeline", "extract"),
     "ingestion": ("pipeline", "extract"),
@@ -228,6 +230,31 @@ def classify_branch(branch: str) -> tuple[str, str]:
     return result
 
 
+# Keep in sync with BRANCH_PREFIX_MAP above.
+_CHANNEL_MAP = {
+    "extract": "telegram",
+    "ingestion": "telegram",
+    "rio": "agent",
+    "theseus": "agent",
+    "astra": "agent",
+    "vida": "agent",
+    "clay": "agent",
+    "leo": "agent",
+    "oberon": "agent",
+    "reweave": "maintenance",
+    "epimetheus": "maintenance",
+    "fix": "maintenance",
+}
+
+
+def classify_source_channel(branch: str, *, github_pr: int = None) -> str:
+    """Derive source_channel from branch prefix and github_pr flag."""
+    if github_pr is not None or branch.startswith("gh-pr-"):
+        return "github"
+    prefix = branch.split("/", 1)[0] if "/" in branch else branch
+    return _CHANNEL_MAP.get(prefix, "unknown")
+
+
 def migrate(conn: sqlite3.Connection):
     """Run schema migrations."""
     conn.executescript(SCHEMA_SQL)
@@ -562,6 +589,34 @@ def migrate(conn: sqlite3.Connection):
         conn.commit()
         logger.info("Migration v21: added github_pr column + index to prs")
 
+    if current < 22:
+        try:
+            conn.execute("ALTER TABLE prs ADD COLUMN source_channel TEXT")
+        except sqlite3.OperationalError:
+            pass
+        conn.execute("""
+            UPDATE prs SET source_channel = CASE
+                WHEN github_pr IS NOT NULL THEN 'github'
+                WHEN branch LIKE 'gh-pr-%%' THEN 'github'
+                WHEN branch LIKE 'theseus/%%' THEN 'agent'
+                WHEN branch LIKE 'rio/%%' THEN 'agent'
+                WHEN branch LIKE 'astra/%%' THEN 'agent'
+                WHEN branch LIKE 'clay/%%' THEN 'agent'
+                WHEN branch LIKE 'vida/%%' THEN 'agent'
+                WHEN branch LIKE 'oberon/%%' THEN 'agent'
+                WHEN branch LIKE 'leo/%%' THEN 'agent'
+                WHEN branch LIKE 'reweave/%%' THEN 'maintenance'
+                WHEN branch LIKE 'epimetheus/%%' THEN 'maintenance'
+                WHEN branch LIKE 'fix/%%' THEN 'maintenance'
+                WHEN branch LIKE 'extract/%%' THEN 'telegram'
+                WHEN branch LIKE 'ingestion/%%' THEN 'telegram'
+                ELSE 'unknown'
+            END
+            WHERE source_channel IS NULL
+        """)
+        conn.commit()
+        logger.info("Migration v22: added source_channel to prs + backfilled from branch prefix")
+
     if current < SCHEMA_VERSION:
         conn.execute(
             "INSERT OR REPLACE INTO schema_version (version) VALUES (?)",

lib/frontmatter.py

@@ -9,6 +9,15 @@ Extracted from merge.py Phase 6 of decomposition (Ganymede-approved plan).
 import yaml
 
 
+def _yaml_quote(value: str) -> str:
+    """Quote a YAML list value if it contains characters that would break parsing."""
+    s = str(value)
+    if ":" in s or s.startswith(("{", "[", "'", '"', "*", "&", "!", "|", ">")):
+        escaped = s.replace('"', '\\"')
+        return f'"{escaped}"'
+    return s
+
+
 # Edge field names recognized in claim frontmatter.
 # Order matters: serialize_edge_fields writes them in this order when appending new fields.
 REWEAVE_EDGE_FIELDS = ("supports", "challenges", "challenged_by", "depends_on", "related", "reweave_edges")
@@ -101,7 +110,7 @@ def serialize_edge_fields(raw_fm_text: str, merged_edges: dict[str, list]) -> st
             if edges:
                 result_lines.append(f"{matched_field}:")
                 for edge in edges:
-                    result_lines.append(f"- {edge}")
+                    result_lines.append(f"- {_yaml_quote(edge)}")
             # Don't increment i — it's already past the old field
             continue
         else:
@@ -115,7 +124,7 @@ def serialize_edge_fields(raw_fm_text: str, merged_edges: dict[str, list]) -> st
             if edges:
                 result_lines.append(f"{field}:")
                 for edge in edges:
-                    result_lines.append(f"- {edge}")
+                    result_lines.append(f"- {_yaml_quote(edge)}")
 
     return "\n".join(result_lines)
 

lib/github_feedback.py

@@ -125,7 +125,8 @@ async def on_eval_complete(conn, forgejo_pr: int, *, outcome: str, review_text:
     if outcome == "approved":
         body = "**Evaluation: Approved**\n\nYour contribution passed automated review and is queued for merge."
         if review_text:
-            body += f"\n\n<details>\n<summary>Review details</summary>\n\n{review_text[:3000]}\n\n</details>"
+            safe_text = review_text[:3000].replace("</details>", "&lt;/details&gt;")
+            body += f"\n\n<details>\n<summary>Review details</summary>\n\n{safe_text}\n\n</details>"
     elif outcome == "rejected":
         body = "**Evaluation: Changes Requested**\n\n"
         if issues:
@@ -133,7 +134,8 @@ async def on_eval_complete(conn, forgejo_pr: int, *, outcome: str, review_text:
             for issue in issues:
                 body += f"- {issue}\n"
         if review_text:
-            body += f"\n<details>\n<summary>Full review</summary>\n\n{review_text[:3000]}\n\n</details>"
+            safe_text = review_text[:3000].replace("</details>", "&lt;/details&gt;")
+            body += f"\n<details>\n<summary>Full review</summary>\n\n{safe_text}\n\n</details>"
         body += (
             "\n\nThe pipeline will attempt automated fixes where possible. "
             "If fixes fail, the PR will be closed — you're welcome to resubmit."

lib/merge.py

@@ -20,7 +20,7 @@ import shutil
 from collections import defaultdict
 
 from . import config, db
-from .db import classify_branch
+from .db import classify_branch, classify_source_channel
 from .contributor import record_contributor_attribution
 from .dedup import dedup_evidence_blocks
 from .domains import detect_domain_from_branch
@@ -117,6 +117,7 @@ async def discover_external_prs(conn) -> int:
                 priority = "high" if origin == "human" else None
                 domain = None if not is_pipeline else detect_domain_from_branch(pr["head"]["ref"])
                 agent, commit_type = classify_branch(pr["head"]["ref"])
+                source_channel = classify_source_channel(pr["head"]["ref"])
 
                 # For human PRs, submitted_by is the Forgejo author.
                 # For pipeline PRs, submitted_by is set later by extract.py (from source proposed_by).
@@ -125,9 +126,9 @@ async def discover_external_prs(conn) -> int:
                 conn.execute(
                     """INSERT OR IGNORE INTO prs
                        (number, branch, status, origin, priority, domain, agent, commit_type,
-                        prompt_version, pipeline_version, submitted_by)
-                       VALUES (?, ?, 'open', ?, ?, ?, ?, ?, ?, ?, ?)""",
-                    (pr["number"], pr["head"]["ref"], origin, priority, domain, agent, commit_type, config.PROMPT_VERSION, config.PIPELINE_VERSION, submitted_by),
+                        prompt_version, pipeline_version, submitted_by, source_channel)
+                       VALUES (?, ?, 'open', ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                    (pr["number"], pr["head"]["ref"], origin, priority, domain, agent, commit_type, config.PROMPT_VERSION, config.PIPELINE_VERSION, submitted_by, source_channel),
                 )
                 db.audit(
                     conn,
@@ -507,9 +508,8 @@ async def _merge_reweave_pr(branch: str) -> tuple[bool, str]:
             branch_fm, _branch_raw_fm, branch_body = parse_yaml_frontmatter(branch_content)
 
             if main_fm is None or branch_fm is None:
-                # Parse failure = something unexpected. Fail the merge, don't fallback
-                # to cherry-pick. (Theseus: loud failure, not silent retry)
-                return False, f"frontmatter parse failed on {fpath} — manual review needed"
+                logger.warning("Reweave merge: frontmatter parse failed on %s — skipping file, continuing PR", fpath)
+                continue
 
             # Superset assertion + merge in one pass.
             # Reweave only adds edges. If branch is missing an edge that main has,

reweave.py

@@ -535,8 +535,9 @@ def _write_edge_regex(neighbor_path: Path, fm_text: str, body_text: str,
     field_re = re.compile(rf"^{edge_type}:\s*$", re.MULTILINE)
     inline_re = re.compile(rf'^{edge_type}:\s*\[', re.MULTILINE)
 
-    entry_line = f'- {orphan_title}'
-    rw_line = f'- {orphan_title}|{edge_type}|{date_str}'
+    from lib.frontmatter import _yaml_quote
+    entry_line = f'- {_yaml_quote(orphan_title)}'
+    rw_line = f'- {_yaml_quote(orphan_title + "|" + edge_type + "|" + date_str)}'
 
     if field_re.search(fm_text):
         # Multi-line list exists — find end of list, append

teleo-pipeline.py

@@ -47,12 +47,21 @@ async def stage_loop(name: str, interval: int, func, conn, breaker: CircuitBreak
                 workers = breaker.max_workers()
                 succeeded, failed = await func(conn, max_workers=workers)
                 if failed > 0 and succeeded == 0:
-                    breaker.record_failure()
+                    try:
+                        breaker.record_failure()
+                    except Exception:
+                        logger.warning("Stage %s: breaker write failed", name)
                 elif succeeded > 0:
-                    breaker.record_success()
+                    try:
+                        breaker.record_success()
+                    except Exception:
+                        logger.warning("Stage %s: breaker write failed", name)
         except Exception:
             logger.exception("Stage %s: unhandled error in cycle", name)
-            breaker.record_failure()
+            try:
+                breaker.record_failure()
+            except Exception:
+                logger.warning("Stage %s: breaker write failed", name)
 
         # Wait for interval or shutdown, whichever comes first
         try: