Merge pull request 'feat(watchdog): auto-escalation של zombie-loops ל-chair — סגירת הלולאה (#218+#222)' (#408) from worktree-zombie-auto-escalate into main

מרכיב את הפרוסות שמוזגו לכלל closed-loop: watchdog שרת-צדי (_zombie_escalation_loop
ב-lifespan, כל 15 דק') שגוזר בריאות per-issue (#222) ומסלים כל zombie מתמשך
(לולאת-recovery, ≥ZOMBIE_ESCALATE_MIN_RECOVERY=2 יקיצות-שחזור) ל-chair דרך
escalate_issue הloop-safe (#218). הטלמטריה agent.escalated (#219) נדלקת מה-Port.

- מדוע שרת-צדי: סוכן תקוע-בלולאה לא יכול להסלים-עצמו אמין; ה-watchdog (מקביל
  ל-Deacon של Gastown, אבל דרך ה-reaper הקיים ולא tier חדש) עושה זאת.
- אידמפוטנטי מבנייה: issue מוסלם הופך agent-unowned → לא zombie בsweep הבא.
- החלטת-ההסלמה (סף+severity) = helper טהור zombie_escalation ב-agent_health (testable).
  severity: ≥5 יקיצות→high, אחרת→medium.
- kill-switch: AGENT_ZOMBIE_AUTO_ESCALATE=0 להשבתה (הבריאות עדיין נצפית, escalate
  ידני עדיין זמין ב-endpoint). ברירת-מחדל ON — הלולאות שורפות budget והתרופה מוכחת.
- 4 בדיקות policy חדשות (11 סה"כ). dry-run חי: 6 issues→0 hzombies→0 הסלמות
  (blast-radius מינימלי על deploy).

Invariants: מקיים G2 (מרחיב את מנגנון-ה-reaper הקיים, לא sweep מקביל; escalate דרך
המסלול הקנוני), G12 (הכל ב-web/, המגע מהשער), §6 (כשל-sweep נלכד ולא-קטלני).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.claude/agents/legal-ceo.md

@@ -226,7 +226,7 @@ Paperclip חוסם אוטומטית כל issue ב-`in_progress` שאין לו ru
 ### שלב 0: בדוק למה התעוררת
 
 **לפני כל דבר אחר** — בדוק את סיבת ההתעוררות (`$PAPERCLIP_WAKE_REASON`):
-- **פעולות סטרוקטורליות (כפתורי-UI, בלי פענוח-טקסט):** אם `$PAPERCLIP_WAKE_PAYLOAD_JSON` מכיל שדה `action` → נתב דטרמיניסטית לפי הערך: `action == "interim_draft"` → **שלב H** (טיוטת ביניים); `action == "party_claims_summary"` → **שלב H2** (סיכום מנהלים). אלו side-quests — אל תסרוק תיקים אחרים, טפל רק בתיק שב-payload (`case_number`/`issueId`).
+- **פעולות סטרוקטורליות (כפתורי-UI, בלי פענוח-טקסט):** אם `$PAPERCLIP_WAKE_PAYLOAD_JSON` מכיל שדה `action` → נתב דטרמיניסטית לפי הערך: `action == "interim_draft"` → **שלב H** (טיוטת ביניים); `action == "party_claims_summary"` → **שלב H2** (סיכום מנהלים). אלו side-quests — אל תסרוק תיקים אחרים, טפל רק בתיק שב-payload (`case_number`/`issueId`). ה-`issueId` שב-payload הוא **issue-ילד ייעודי שמשויך אליך** (נוצר ע"י המערכת כדי שהריצה לא תבוטל כשה-issue הראשי ממתין-ליו"ר; #227) — הרץ את הפעולה, פרסם comment עם התוצאה, ו**סגור את ה-issue-ילד הזה כ-done** בסיום. אל תיגע ב-issue הראשי של התיק.
 - אם ה-reason מכיל `user_commented` **ו**-`$PAPERCLIP_WAKE_PAYLOAD_JSON` כולל `issueId` → **דלג ישירות לסעיף "טיפול בתגובות חדשות מחיים"** עם ה-issue הזה. אל תסרוק תיקים אחרים. **טפל רק בתגובה.**
 - אם ה-reason מכיל `agent_completion` → דלג לשלב E/F בהתאם לסוכן שסיים
 - אם ה-reason מכיל `precedent_extraction_` → **דלג לסעיף "חילוץ פסיקה אוטומטי"**. אל תיגע בתיקים — זו עבודת ספרייה.
@@ -719,13 +719,13 @@ ls data/cases/$CASE_NUMBER/documents/research/analysis-and-research.md
    ```
    mcp__legal-ai__export_interim_draft(case_number="...")
    ```
-   מייצר `data/cases/{case_number}/exports/טיוטת-ביניים-v{N}.docx`, מעדכן `active_draft_path`.
+   מייצר `data/cases/{case_number}/exports/טיוטה-טענות_הצדדים_{N}.docx`, מעדכן `active_draft_path`.
 
 5. **דווח לחיים** (כולל מייל דרך `scripts/notify.py`):
    ```
    ## טיוטת ביניים מוכנה — ערר {case_number}
 
-   📄 **קובץ:** `data/cases/{case_number}/exports/טיוטת-ביניים-v{N}.docx`
+   📄 **קובץ:** `data/cases/{case_number}/exports/טיוטה-טענות_הצדדים_{N}.docx`
 
    ### מה כלול
    | בלוק | כותרת | מילים |

mcp-server/src/legal_mcp/services/argument_aggregator.py

@@ -221,7 +221,7 @@ async def aggregate_claims_to_arguments(
 
         # Pull all claims for this case, grouped by party.
         rows = await conn.fetch(
-            """SELECT id, party_role, claim_text, claim_index, source_document
+            """SELECT id, party_role, claim_text, claim_index, source_document, party_name
                FROM claims
                WHERE case_id = $1
                ORDER BY party_role, claim_index""",
@@ -249,10 +249,13 @@ async def aggregate_claims_to_arguments(
         # Map deprecated 'appraiser' or unknown labels to 'unknown'.
         if party not in ALLOWED_PARTIES:
             party = "unknown"
-        party_name = (
+        # Prefer the stored party_name (stamped by the extractor, #224); fall
-            (r["source_document"] or "").strip()
+        # back to source_document for legacy claims predating the stamping.
-            if party in SPLIT_PARTIES else ""
+        # Single-voice sides (appellant/committee) stay ''.
-        )
+        if party in SPLIT_PARTIES:
+            party_name = (r["party_name"] or "").strip() or (r["source_document"] or "").strip()
+        else:
+            party_name = ""
         by_group.setdefault((party, party_name), []).append(dict(r))
 
     # Valid claim_ids for this case == the ids of the claims we just fetched.

mcp-server/src/legal_mcp/services/block_writer.py

@@ -192,16 +192,16 @@ BLOCK_PROMPTS = {
 ## כללים קריטיים:
 - **סנתז טענות דומות** — אל תרשום כל טענה בנפרד. קבץ טענות דומות לנושא אחד. למשל: כל הטענות על הודעות → סעיף אחד, כל הטענות על רכוש משותף → סעיף אחד.
 - גוף שלישי: "העוררים טוענים כי...", "הוועדה המקומית ציינה כי..."
-- **מבנה קבוע עם 3 חלקים:**
+- **מבנה: חלק (סעיף עם כותרת-משנה) לכל צד שמופיע בטענות שחולצו למטה** — לפי הכותרות שם (### ...). אל תמציא צדדים שאינם מופיעים ואל תשמיט צד שמופיע.
-  1. "טענות העוררים" — 8-12 סעיפים מקובצים לפי נושא
+  - "טענות העוררים" — 8-12 סעיפים מקובצים לפי נושא.
-  2. "עמדת הוועדה המקומית" — 5-8 סעיפים
+  - **צד-משיב/מתנגד: כתוב חלק נפרד לכל כתב-תשובה** שמופיע (למשל "עמדת משיבות 2-3" ו-"עמדת משיבים 4-6" בנפרד) — 5-10 סעיפים לכל אחד. **אל תמזג משיבים שונים לסעיף אחד**, ואם עמדותיהם מנוגדות — שקף זאת.
-  3. "עמדת מבקשי ההיתר" (אם יש) — 5-10 סעיפים
+  - "עמדת הוועדה המקומית" / "עמדת מבקשי ההיתר" (אם מופיעים) — 5-8 סעיפים.
 - כותרת: "תמצית טענות הצדדים"
 - נאמנות למקור — לא להמציא טענות, אבל כן לאחד ולסכם טענות חוזרות
 - אין ניתוח, אין מסקנות, אין הערכה ("טענה חלשה/חזקה")
 - רק מכתבי טענות מקוריים (לא השלמות טיעון)
 - מספור רציף
-- **יעד אורך: 800-1500 מילים**
+- **יעד אורך: 800-1500 מילים** (יותר כשיש כמה כתבי-תשובה נפרדים — כל צד מקבל את מלוא ההתייחסות)
 
 ## טענות שחולצו (קבץ טענות דומות לנושאים):
 {claims_context}
@@ -284,6 +284,7 @@ BLOCK_PROMPTS = {
 - **ללא כפילות** — הפנה לבלוקים קודמים: "כאמור בסעיף X לעיל"
 - **מספור רציף** — המשך מספור מהבלוק הקודם
 - מותרות כותרות-משנה כשיש נושאים נפרדים לחלוטין
+- **צדדים מרובים** — כשיש כמה משיבים/מתנגדים או כתבי-תשובה נפרדים (ראה הכותרות שתחת "טענות" לעיל, למשל "משיבות 2-3" מול "משיבים 4-6"), התייחס לעמדת כל אחד לגופה ואל תמזג אותם ל"טענות המשיבים" גורפות. אם משיבים שונים נוקטים **עמדות מנוגדות זו לזו** — ציין זאת במפורש והכרע ביניהן בנימוק; אם עמדותיהם משלימות — ניתן לאגד תוך שמירת הייחוס.
 
 ## כיוון מאושר (חובה):
 {direction_context}
@@ -629,18 +630,49 @@ async def _build_claims_context(case_id: UUID) -> str:
         # Fallback to all non-block-zayin claims if no source claims exist.
         source_claims = [c for c in claims if c.get("source_document", "") != "block-zayin"] or claims
 
-    lines = []
+    # Group by (party_role, brief) so a multi-litigant side (respondent /
-    current_role = ""
+    # permit_applicant) whose claims come from distinct pleadings reads as
+    # SEPARATE positions — e.g. "כתב תשובה משיבות 2-3" vs "משיבים 4-6" — instead
+    # of one merged "טענות המשיבים" (#224). The brief label is the source
+    # pleading, and which sides split is SPLIT_PARTIES — the SAME rule the
+    # argument aggregator uses, so block ז and the arguments panel agree (G2).
+    from legal_mcp.services.argument_aggregator import SPLIT_PARTIES
+
     role_heb = {"appellant": "טענות העוררים", "respondent": "טענות המשיבים",
                 "committee": "עמדת הוועדה המקומית", "permit_applicant": "עמדת מבקשי ההיתר"}
-    claim_num = 0
+    role_order = ["appellant", "committee", "respondent", "permit_applicant"]
+
+    groups: dict[tuple[str, str], list[dict]] = {}
     for c in source_claims:
-        if c["party_role"] != current_role:
+        role = c.get("party_role", "") or ""
-            current_role = c["party_role"]
+        # Prefer the stored party_name (stamped by the extractor, #224); fall
-            lines.append(f"\n### {role_heb.get(current_role, current_role)}")
+        # back to source_document for legacy claims predating the stamping.
-        claim_num += 1
+        if role in SPLIT_PARTIES:
-        lines.append(f"טענה #{claim_num}: {c['claim_text'][:400]}")
+            brief = (c.get("party_name") or "").strip() or (c.get("source_document") or "").strip()
-    lines.append(f"\n**סה\"כ {claim_num} טענות. ענה על כל טענה מהותית; טענות [bundle] — אגד; טענות [skip] — ציון קצר בלבד.**")
+        else:
+            brief = ""
+        groups.setdefault((role, brief), []).append(c)
+
+    def _sort_key(k: tuple[str, str]) -> tuple[int, str]:
+        role, brief = k
+        idx = role_order.index(role) if role in role_order else len(role_order)
+        return (idx, brief)
+
+    lines: list[str] = []
+    claim_num = 0
+    for role, brief in sorted(groups.keys(), key=_sort_key):
+        header = role_heb.get(role, role or "טענות נוספות")
+        if brief:
+            header = f"{header} — {brief}"
+        lines.append(f"\n### {header}")
+        for c in groups[(role, brief)]:
+            claim_num += 1
+            lines.append(f"טענה #{claim_num}: {c['claim_text'][:400]}")
+    lines.append(
+        f"\n**סה\"כ {claim_num} טענות. שמור על ההפרדה בין הצדדים לעיל — "
+        f"לכל צד/כתב-תשובה עמדה נפרדת (אל תמזג משיבים שונים). ענה על כל טענה "
+        f"מהותית; טענות [bundle] — אגד; טענות [skip] — ציון קצר בלבד.**"
+    )
     return "\n".join(lines)
 
 

mcp-server/src/legal_mcp/services/claims_extractor.py

@@ -368,8 +368,15 @@ async def extract_and_store_claims(
 
     # Determine claim_type from document type and title
     claim_type = _infer_claim_type(doc_type, source_name)
+    # Stamp party_name at the source (#224). For a multi-litigant side
+    # (respondent / permit_applicant) the brief label IS the source pleading, so
+    # opposing briefs (משיבות 2-3 vs משיבים 4-6) stay distinct downstream without
+    # the aggregator/block-writer having to re-derive it. SPLIT_PARTIES is the
+    # single rule shared with the aggregator (G2); single-voice sides stay ''.
+    from legal_mcp.services.argument_aggregator import SPLIT_PARTIES
     for c in claims:
         c["claim_type"] = claim_type
+        c["party_name"] = source_name if c.get("party_role") in SPLIT_PARTIES else ""
 
     stored = await db.store_claims(case_id, claims, source_document=source_name)
     # Mark this document analysed (WS2 / #201). store_claims already replaced

mcp-server/src/legal_mcp/services/db.py

@@ -1900,6 +1900,38 @@ CREATE INDEX IF NOT EXISTS idx_legal_arguments_party_name
 """
 
 
+# V51 (#226): hearing-attendance provenance for the "מה קרה בדיון" panel. The
+# ערר-hearing protocol names who actually appeared (עוררים/משיבים + their
+# counsel) and, when stated, the presiding panel. ``_extract_header`` already
+# extracts this feed but only ``hearing_date`` had a canonical home (on cases);
+# the attendee lists were returned as provenance and dropped. This column gives
+# them a home so the panel can display them.
+#
+# This is DISTINCT from ``decisions.panel_members`` (G2, not a parallel path):
+# that column is the authoring tribunal recorded on a specific written-decision
+# version (the DOCX signature block). ``hearing_attendees`` is a snapshot of who
+# was present at the *hearing*, extracted from the protocol — hearing-event
+# provenance, not decision-authorship. Shape:
+#   {"panel_members": [...], "appellants_present": [...], "respondents_present": [...]}
+SCHEMA_V51_SQL = """
+ALTER TABLE cases ADD COLUMN IF NOT EXISTS hearing_attendees JSONB NOT NULL DEFAULT '{}';
+"""
+
+
+# V52 (#224): backfill claims.party_name for existing rows. The extractor now
+# stamps party_name = source_document for the multi-litigant sides at write time
+# (claims_extractor), but claims stored before that shipped have party_name=''.
+# For those, the brief label is the source pleading — same rule the aggregator
+# uses (respondent / permit_applicant split). Idempotent: only fills empties, so
+# re-running is a no-op. Single-voice sides (appellant/committee) stay ''.
+SCHEMA_V52_SQL = """
+UPDATE claims SET party_name = source_document
+WHERE party_role IN ('respondent', 'permit_applicant')
+  AND COALESCE(party_name, '') = ''
+  AND COALESCE(source_document, '') <> '';
+"""
+
+
 # Stable, arbitrary key for the session-level advisory lock that serialises
 # schema DDL across processes. Every short-lived process (cron drains, services)
 # re-runs the idempotent migrations on startup; without this lock two processes
@@ -1972,6 +2004,8 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V48_SQL)
         await conn.execute(SCHEMA_V49_SQL)
         await conn.execute(SCHEMA_V50_SQL)
+        await conn.execute(SCHEMA_V51_SQL)
+        await conn.execute(SCHEMA_V52_SQL)
 
 
 async def init_schema() -> None:
@@ -2194,7 +2228,7 @@ async def update_case(case_id: UUID, **fields) -> dict | None:
     set_clauses = []
     values = []
     for i, (key, val) in enumerate(fields.items(), start=2):
-        if key in ("appellants", "respondents", "tags"):
+        if key in ("appellants", "respondents", "tags", "hearing_attendees"):
             val = json.dumps(val)
         set_clauses.append(f"{key} = ${i}")
         values.append(val)
@@ -2207,7 +2241,7 @@ async def update_case(case_id: UUID, **fields) -> dict | None:
 
 def _row_to_case(row: asyncpg.Record) -> dict:
     d = dict(row)
-    for field in ("appellants", "respondents", "tags"):
+    for field in ("appellants", "respondents", "tags", "hearing_attendees"):
         if isinstance(d.get(field), str):
             d[field] = json.loads(d[field])
     d["id"] = str(d["id"])

mcp-server/src/legal_mcp/services/docx_exporter.py

@@ -399,8 +399,15 @@ _INTERIM_BLOCK_ORDER = [
 ]
 
 
-def _draft_filename_prefix(mode: str) -> str:
+def _draft_naming(mode: str) -> tuple[str, str]:
-    return "טיוטת-ביניים" if mode == "interim" else "טיוטה"
+    """(filename prefix, version separator) per export mode.
+
+    interim → ``טיוטה-טענות_הצדדים_{N}.docx`` (chair-requested naming);
+    final   → ``טיוטה-v{N}.docx``.
+    """
+    if mode == "interim":
+        return "טיוטה-טענות_הצדדים", "_"
+    return "טיוטה", "-v"
 
 
 async def export_decision(
@@ -485,16 +492,19 @@ async def export_decision(
     if not output_path:
         export_dir = config.find_case_dir(case["case_number"]) / "exports"
         export_dir.mkdir(parents=True, exist_ok=True)
-        prefix = _draft_filename_prefix(mode)
+        prefix, sep = _draft_naming(mode)
-        existing = sorted(export_dir.glob(f"{prefix}-v*.docx"))
+        existing = sorted(export_dir.glob(f"{prefix}{sep}*.docx"))
         next_ver = 1
         for p in existing:
             try:
-                ver = int(p.stem.split("-v")[1])
+                # Version is the trailing integer after the separator. Using
+                # rsplit keeps this correct even when the prefix itself contains
+                # the separator char (e.g. "טיוטה-טענות_הצדדים" with sep="_").
+                ver = int(p.stem.rsplit(sep, 1)[1])
                 next_ver = max(next_ver, ver + 1)
             except (IndexError, ValueError):
                 pass
-        output_path = str(export_dir / f"{prefix}-v{next_ver}.docx")
+        output_path = str(export_dir / f"{prefix}{sep}{next_ver}.docx")
 
     # Persist through the storage layer (INV-STG1). Under the filesystem
     # backend the bytes land at output_path exactly as before; a caller-

mcp-server/src/legal_mcp/services/protocol_analyzer.py

@@ -259,13 +259,25 @@ async def _extract_header(protocol_text: str, case_id: UUID) -> dict:
         except ValueError:
             logger.info("protocol header: unparseable hearing_date %r", hearing_date)
 
+    # Attendees → cases.hearing_attendees (#226). Unlike hearing_date there is no
+    # chair-entry path to protect, and re-running to point at the correct ערר
+    # protocol (#223) must refresh who appeared — so write whenever the operative
+    # protocol yielded any names, but never clobber good data with an empty
+    # extraction (all-empty feed → leave the prior snapshot intact).
+    attendees = {
+        "panel_members": feed.get("panel_members") or [],
+        "appellants_present": feed.get("appellants_present") or [],
+        "respondents_present": feed.get("respondents_present") or [],
+    }
+    if any(attendees.values()):
+        await db.update_case(case_id, hearing_attendees=attendees)
+        applied["hearing_attendees"] = True
+
     return {
         "status": "ok",
         "feed": {
             "hearing_date": hearing_date,
-            "panel_members": feed.get("panel_members") or [],
+            **attendees,
-            "appellants_present": feed.get("appellants_present") or [],
-            "respondents_present": feed.get("respondents_present") or [],
         },
         "applied_to_case": applied,
     }

mcp-server/src/legal_mcp/tools/drafting.py

@@ -716,7 +716,7 @@ async def write_interim_draft(case_number: str, instructions: str = "") -> str:
 async def export_interim_draft(case_number: str, output_path: str = "") -> str:
     """ייצוא טיוטת ביניים ל-DOCX — אותו עיצוב של טיוטה רגילה (David, RTL,
     bookmarks), אבל בסדר חדש: רקע → תכניות+היתרים → טענות → הליכים, ללא
-    דיון/סיכום/חתימות. שם הקובץ: טיוטת-ביניים-v{N}.docx.
+    דיון/סיכום/חתימות. שם הקובץ: טיוטה-טענות_הצדדים_{N}.docx.
 
     Args:
         case_number: מספר תיק הערר

web-ui/src/app/cases/[caseNumber]/page.tsx

@@ -17,6 +17,7 @@ import { DraftsPanel } from "@/components/cases/drafts-panel";
 import { DecisionBlocksPanel } from "@/components/cases/decision-blocks-panel";
 import { LegalArgumentsPanel } from "@/components/cases/legal-arguments-panel";
 import { PositionsPanel } from "@/components/cases/positions-panel";
+import { HearingChangesPanel } from "@/components/cases/hearing-changes-panel";
 import { CitationVerificationPanel } from "@/components/compose/citation-verification-panel";
 import { AgentActivityFeed } from "@/components/cases/agent-activity-feed";
 import { AgentActivityPreview } from "@/components/cases/agent-activity-preview";
@@ -156,11 +157,28 @@ export default function CaseDetailPage({
               issues, with the aggregated by-party arguments collapsible below
               (merged from the deleted /compose editor, mockup 18f). */}
           <TabsContent value="arguments" className="mt-0 space-y-4">
-            <Card className="bg-surface border-rule shadow-sm">
+            {/* עמדות וטענות — collapsed by default (chair preference #226),
-              <CardContent className="px-6 py-5">
+                mirroring the two accordions below. */}
-                <PositionsPanel caseNumber={caseNumber} />
+            <Accordion type="single" collapsible>
-              </CardContent>
+              <AccordionItem
-            </Card>
+                value="positions"
+                className="overflow-hidden rounded-xl border border-rule bg-surface shadow-sm"
+              >
+                <AccordionTrigger className="px-6 py-4 hover:no-underline">
+                  <span className="flex flex-1 flex-col items-start">
+                    <span className="text-navy text-[0.95rem] font-bold">
+                      עמדות וטענות
+                    </span>
+                    <span className="text-ink-muted text-[0.78rem]">
+                      סוגיות-המחלוקת ועמדות הצדדים · עורך עמדת-היו״ר מזין את בלוק י׳
+                    </span>
+                  </span>
+                </AccordionTrigger>
+                <AccordionContent className="px-6 pb-5 pt-0">
+                  <PositionsPanel caseNumber={caseNumber} />
+                </AccordionContent>
+              </AccordionItem>
+            </Accordion>
             <Accordion type="single" collapsible>
               <AccordionItem
                 value="byparty"
@@ -181,6 +199,30 @@ export default function CaseDetailPage({
                 </AccordionContent>
               </AccordionItem>
             </Accordion>
+            {/* מה קרה בדיון — comparative analysis of the hearing protocol vs.
+                the written pleadings (#226). Sits below the aggregated arguments
+                because it speaks to those same arguments in their oral gloss.
+                Collapsed by default, mirroring the "by-party" accordion above. */}
+            <Accordion type="single" collapsible>
+              <AccordionItem
+                value="hearing"
+                className="overflow-hidden rounded-xl border border-rule bg-surface shadow-sm"
+              >
+                <AccordionTrigger className="px-6 py-4 hover:no-underline">
+                  <span className="flex flex-1 flex-col items-start">
+                    <span className="text-navy text-[0.95rem] font-bold">
+                      מה קרה בדיון
+                    </span>
+                    <span className="text-ink-muted text-[0.78rem]">
+                      השוואת הטענות בכתב מול הדיון בעל-פה — מה התחזק ומה נטען לראשונה
+                    </span>
+                  </span>
+                </AccordionTrigger>
+                <AccordionContent className="px-6 pb-5 pt-0">
+                  <HearingChangesPanel caseNumber={caseNumber} />
+                </AccordionContent>
+              </AccordionItem>
+            </Accordion>
           </TabsContent>
 
           {/* אימות פסיקה — per-argument supporting-precedent verify gate (#154),

web-ui/src/components/cases/drafts-panel.tsx

@@ -31,7 +31,9 @@ import {
   usePartyClaimsSummary,
   useGeneratePartyClaimsSummary,
   useGenerateInterimDraft,
+  useGenerateStatus,
 } from "@/lib/api/generate";
+import { GenerateStatusChip } from "@/components/cases/generate-status-chip";
 import {
   useCaseFeedback,
   useCreateFeedback,
@@ -89,7 +91,7 @@ function formatDate(epoch: number): string {
 /**
  * Pick the newest export file whose name starts with `prefix`, preferring the
  * highest v-number and falling back to the latest created_at. Used by the
- * "הפקת מסמכים" card to surface the most recent טיוטת-ביניים-* file (#214).
+ * "הפקת מסמכים" card to surface the most recent טיוטה-טענות_הצדדים_* file (#214).
  */
 function pickLatestVersioned(
   files: ExportFile[] | undefined,
@@ -98,7 +100,10 @@ function pickLatestVersioned(
   const matches = (files ?? []).filter((f) => f.filename.startsWith(prefix));
   if (!matches.length) return null;
   const withVer = matches.map((f) => {
-    const m = f.filename.match(/v(\d+)/);
+    // Version is the trailing integer, after either "-v" (final) or "_"
+    // (interim: טיוטה-טענות_הצדדים_{N}). Anchor to the end so the "_" inside
+    // the interim prefix isn't mistaken for the version separator.
+    const m = f.filename.match(/[_v](\d+)(?:\.\w+)?$/);
     return { ...f, version: m ? parseInt(m[1], 10) : null };
   });
   withVer.sort((a, b) => {
@@ -133,6 +138,9 @@ export function DraftsPanel({
   const { data: summary } = usePartyClaimsSummary(caseNumber);
   const genSummary = useGeneratePartyClaimsSummary(caseNumber);
   const genInterim = useGenerateInterimDraft(caseNumber);
+  // Server-derived background-run status (#227 ג) — survives navigation/reload.
+  const summaryStatus = useGenerateStatus(caseNumber, "party_claims_summary");
+  const interimStatus = useGenerateStatus(caseNumber, "interim_draft");
   const qc = useQueryClient();
 
   const fileRef = useRef<HTMLInputElement>(null);
@@ -167,8 +175,8 @@ export function DraftsPanel({
 
   // ── "הפקת מסמכים" derived state (#214) ──
   // Latest interim ("party-claims") partial draft in the exports table — the poll
-  // target for button 2. The CEO writes it as טיוטת-ביניים-{case}-vN.docx.
+  // target for button 2. The CEO writes it as טיוטה-טענות_הצדדים_{N}.docx.
-  const latestInterim = pickLatestVersioned(exports, "טיוטת-ביניים-");
+  const latestInterim = pickLatestVersioned(exports, "טיוטה-טענות_הצדדים_");
   const summaryReady = Boolean(summary?.markdown);
 
   function handleGenerateSummary() {
@@ -342,6 +350,11 @@ export function DraftsPanel({
                     "הפק"
                   )}
                 </Button>
+                <GenerateStatusChip
+                  status={summaryStatus.data}
+                  onRetry={handleGenerateSummary}
+                  retrying={genSummary.isPending}
+                />
                 {summaryReady && (
                   <>
                     <Button
@@ -410,6 +423,11 @@ export function DraftsPanel({
                     "הפק"
                   )}
                 </Button>
+                <GenerateStatusChip
+                  status={interimStatus.data}
+                  onRetry={handleGenerateInterim}
+                  retrying={genInterim.isPending}
+                />
                 {latestInterim && (
                   <span className="text-[0.7rem] text-ink-muted">
                     {latestInterim.version

web-ui/src/components/cases/generate-status-chip.tsx

@@ -0,0 +1,127 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { Button } from "@/components/ui/button";
+import { CheckCircle2, Clock, Loader2, RotateCw, XCircle } from "lucide-react";
+import type { GenerateStatus } from "@/lib/api/generate";
+
+/**
+ * Background-run status chip for the "הפקת מסמכים" generation actions (#227 ג).
+ *
+ * The chip is driven entirely by the server-derived {@link GenerateStatus} — it
+ * holds no run state of its own, so it renders correctly after the chair
+ * navigates away and back. The running timer is anchored to the server
+ * `started_at`, not a click-time counter, so the elapsed value is right on
+ * return too.
+ */
+
+function useElapsedSeconds(startedAt: string | null, active: boolean): number | null {
+  const [now, setNow] = useState(() => Date.now());
+  useEffect(() => {
+    if (!active) return;
+    const id = setInterval(() => setNow(Date.now()), 1_000);
+    return () => clearInterval(id);
+  }, [active]);
+  if (!startedAt) return null;
+  const start = Date.parse(startedAt);
+  if (Number.isNaN(start)) return null;
+  return Math.max(0, Math.floor((now - start) / 1000));
+}
+
+function fmtDuration(sec: number): string {
+  const m = Math.floor(sec / 60);
+  const s = sec % 60;
+  if (m >= 60) {
+    const h = Math.floor(m / 60);
+    return `${h}:${String(m % 60).padStart(2, "0")}:${String(s).padStart(2, "0")}`;
+  }
+  return `${m}:${String(s).padStart(2, "0")}`;
+}
+
+function relativeTime(iso: string | null): string {
+  if (!iso) return "";
+  const t = Date.parse(iso);
+  if (Number.isNaN(t)) return "";
+  const sec = Math.max(0, Math.floor((Date.now() - t) / 1000));
+  if (sec < 60) return "הרגע";
+  const min = Math.floor(sec / 60);
+  if (min < 60) return `לפני ${min} דק׳`;
+  const hr = Math.floor(min / 60);
+  if (hr < 24) return `לפני ${hr} שע׳`;
+  return `לפני ${Math.floor(hr / 24)} ימים`;
+}
+
+const CHIP_BASE =
+  "inline-flex items-center gap-1.5 rounded-full border px-2.5 py-1 text-xs font-semibold whitespace-nowrap";
+
+export function GenerateStatusChip({
+  status,
+  onRetry,
+  retrying,
+}: {
+  status: GenerateStatus | undefined;
+  onRetry?: () => void;
+  retrying?: boolean;
+}) {
+  const state = status?.state ?? "idle";
+  const elapsed = useElapsedSeconds(status?.started_at ?? null, state === "running");
+
+  if (!status || state === "idle") return null;
+
+  if (state === "queued") {
+    return (
+      <span className={`${CHIP_BASE} bg-rule-soft text-ink-soft border-rule`}>
+        <Clock className="size-3.5" aria-hidden />
+        בתור…
+      </span>
+    );
+  }
+
+  if (state === "running") {
+    return (
+      <span className={`${CHIP_BASE} bg-info-bg text-info border-info/35`}>
+        <Loader2 className="size-3.5 animate-spin" aria-hidden />
+        רץ ברקע
+        {elapsed !== null && (
+          <span className="tabular-nums">· {fmtDuration(elapsed)}</span>
+        )}
+      </span>
+    );
+  }
+
+  if (state === "done") {
+    const rel = relativeTime(status.finished_at);
+    return (
+      <span className={`${CHIP_BASE} bg-success-bg text-success border-success/35`}>
+        <CheckCircle2 className="size-3.5" aria-hidden />
+        הושלם{rel && ` · ${rel}`}
+      </span>
+    );
+  }
+
+  // failed
+  return (
+    <span className="inline-flex items-center gap-2">
+      <span className={`${CHIP_BASE} bg-danger-bg text-danger border-danger/35`}>
+        <XCircle className="size-3.5" aria-hidden />
+        {status.detail || "נכשל"}
+      </span>
+      {onRetry && (
+        <Button
+          variant="ghost"
+          size="sm"
+          className="h-7 px-2 text-gold-deep"
+          disabled={retrying}
+          onClick={onRetry}
+        >
+          {retrying ? (
+            <Loader2 className="size-3.5 animate-spin me-1" />
+          ) : (
+            <RotateCw className="size-3.5 me-1" />
+          )}
+          נסה שוב
+        </Button>
+      )}
+    </span>
+  );
+}

web-ui/src/components/cases/hearing-changes-panel.tsx

@@ -0,0 +1,360 @@
+"use client";
+
+import { useMemo, useState } from "react";
+import { Badge } from "@/components/ui/badge";
+import { Button } from "@/components/ui/button";
+import { Skeleton } from "@/components/ui/skeleton";
+import {
+  ArrowDown,
+  ArrowUp,
+  CalendarDays,
+  HelpCircle,
+  Link2,
+  Loader2,
+  Plus,
+  RotateCw,
+  Sparkles,
+} from "lucide-react";
+import { toast } from "sonner";
+import {
+  CHANGE_LABELS_HE,
+  PROTOCOL_PARTY_LABELS_HE,
+  PROTOCOL_PARTY_ORDER,
+  useAnalyzeProtocol,
+  useProtocolAnalysis,
+  type ProtocolAnalysisRow,
+  type ProtocolChangeType,
+  type ProtocolPartyRole,
+} from "@/lib/api/protocol-analysis";
+
+const PARTY_MARK_TONE: Record<ProtocolPartyRole, string> = {
+  appellant: "bg-info",
+  respondent: "bg-gold-deep",
+  committee: "bg-success",
+  permit_applicant: "bg-warn",
+  unknown: "bg-ink-muted",
+  "": "bg-ink-muted",
+};
+
+const CHANGE_BADGE_TONE: Record<ProtocolChangeType, string> = {
+  strengthened: "bg-gold-wash text-gold-deep border-gold/40",
+  newly_raised: "bg-emerald-50 text-emerald-900 border-emerald-200",
+  dropped: "bg-rule-soft text-ink-soft border-rule",
+};
+
+const CHANGE_ACCENT: Record<ProtocolChangeType, string> = {
+  strengthened: "border-s-gold",
+  newly_raised: "border-s-emerald-500",
+  dropped: "border-s-rule",
+};
+
+function ChangeIcon({ type }: { type: ProtocolChangeType }) {
+  if (type === "strengthened") return <ArrowUp className="size-3.5" aria-hidden />;
+  if (type === "newly_raised") return <Plus className="size-3.5" aria-hidden />;
+  return <ArrowDown className="size-3.5" aria-hidden />;
+}
+
+type FilterKey = "all" | ProtocolChangeType;
+
+function AttendeeColumn({
+  label,
+  dotTone,
+  names,
+}: {
+  label: string;
+  dotTone: string;
+  names: string[];
+}) {
+  if (!names.length) return null;
+  return (
+    <div className="min-w-[210px] flex-1">
+      <div className="text-gold-deep mb-1.5 flex items-center gap-1.5 text-[0.7rem] font-bold">
+        <span className={`size-1.5 flex-none rounded-full ${dotTone}`} aria-hidden />
+        {label}
+      </div>
+      <div className="text-ink-soft text-xs leading-relaxed">
+        {names.join(" · ")}
+      </div>
+    </div>
+  );
+}
+
+function DevelopmentCard({ row }: { row: ProtocolAnalysisRow }) {
+  return (
+    <div
+      className={`bg-surface border-rule overflow-hidden rounded-lg border border-s-4 shadow-sm ${CHANGE_ACCENT[row.change_type]}`}
+    >
+      <div className="space-y-2.5 px-4 py-3.5">
+        <div className="flex flex-wrap items-start gap-2.5">
+          <Badge
+            variant="outline"
+            className={`${CHANGE_BADGE_TONE[row.change_type]} flex-none gap-1 text-[0.72rem] font-bold`}
+          >
+            <ChangeIcon type={row.change_type} />
+            {CHANGE_LABELS_HE[row.change_type]}
+          </Badge>
+          <span className="text-navy flex-1 text-sm font-bold leading-snug">
+            {row.argument_title}
+          </span>
+        </div>
+
+        {row.summary && (
+          <p className="text-ink-soft text-[0.82rem] leading-relaxed">
+            {row.summary}
+          </p>
+        )}
+
+        {row.sharpened_question && (
+          <div className="bg-parchment border-rule-soft rounded-md border px-3 py-2">
+            <div className="text-gold-deep mb-1 flex items-center gap-1.5 text-[0.66rem] font-bold">
+              <HelpCircle className="size-3" aria-hidden />
+              השאלה שהתחדדה
+            </div>
+            <div className="text-ink text-xs leading-relaxed">
+              {row.sharpened_question}
+            </div>
+          </div>
+        )}
+
+        {row.evidence_quote && (
+          <blockquote className="border-rule text-ink-muted border-s-[3px] px-3 text-xs italic leading-relaxed">
+            &ldquo;{row.evidence_quote}&rdquo;
+            <span className="text-ink-muted mt-1 block text-[0.68rem] not-italic font-semibold">
+              — מתוך פרוטוקול הדיון
+            </span>
+          </blockquote>
+        )}
+
+        <div className="flex flex-wrap items-center gap-2 pt-0.5">
+          {row.change_type === "newly_raised" ? (
+            <span className="text-emerald-800 inline-flex items-center gap-1.5 text-[0.72rem] font-semibold">
+              <Sparkles className="size-3.5" aria-hidden />
+              אין מקבילה בכתב — עלה לראשונה בדיון
+            </span>
+          ) : row.argument_id ? (
+            <span className="text-ink-muted inline-flex items-center gap-1.5 text-[0.72rem] font-semibold">
+              <Link2 className="size-3.5" aria-hidden />
+              מבוסס על טיעון כתוב קיים
+            </span>
+          ) : null}
+        </div>
+      </div>
+    </div>
+  );
+}
+
+type HearingChangesPanelProps = {
+  caseNumber: string;
+};
+
+export function HearingChangesPanel({ caseNumber }: HearingChangesPanelProps) {
+  const { data, isPending, isError, error } = useProtocolAnalysis(caseNumber);
+  const analyze = useAnalyzeProtocol(caseNumber);
+  const [filter, setFilter] = useState<FilterKey>("all");
+
+  const handleAnalyze = () => {
+    analyze.mutate(undefined, {
+      onSuccess: (res) => {
+        if (res.status === "queued") {
+          toast.success("נשלח למנתח המשפטי — הניתוח ירוץ ברקע; רענן בעוד כמה דקות.");
+        } else {
+          toast.warning(
+            `לא ניתן להריץ אוטומטית (${res.reason}). ניתן להריץ ידנית מ-Claude Code.`,
+          );
+        }
+      },
+      onError: (e) => toast.error(`שגיאה: ${(e as Error).message}`),
+    });
+  };
+
+  const header = data?.header;
+  const counts = data?.by_change ?? {};
+  const strengthened = counts.strengthened ?? 0;
+  const newlyRaised = counts.newly_raised ?? 0;
+  const dropped = counts.dropped ?? 0;
+
+  const sections = useMemo(() => {
+    if (!data) return [];
+    return PROTOCOL_PARTY_ORDER.map((role) => {
+      const rows = (data.by_party[role] ?? []).filter(
+        (r) => filter === "all" || r.change_type === filter,
+      );
+      return { role, rows };
+    }).filter((s) => s.rows.length > 0);
+  }, [data, filter]);
+
+  const runButton = (
+    <Button
+      variant="outline"
+      size="sm"
+      disabled={analyze.isPending}
+      onClick={handleAnalyze}
+    >
+      {analyze.isPending ? (
+        <Loader2 className="me-1.5 size-3.5 animate-spin" />
+      ) : (
+        <RotateCw className="me-1.5 size-3.5" />
+      )}
+      {data?.total ? "נתח מחדש" : "נתח את פרוטוקול הדיון"}
+    </Button>
+  );
+
+  return (
+    <div className="space-y-4">
+      {isPending ? (
+        <div className="space-y-2">
+          <Skeleton className="h-20 w-full" />
+          <Skeleton className="h-24 w-full" />
+        </div>
+      ) : isError ? (
+        <p className="text-danger text-sm">
+          שגיאה בטעינת ניתוח-הפרוטוקול: {(error as Error).message}
+        </p>
+      ) : !data?.total ? (
+        <div className="bg-surface border-rule rounded-lg border border-dashed px-6 py-8 text-center">
+          <h3 className="text-navy text-sm font-bold">טרם נותח פרוטוקול</h3>
+          <p className="text-ink-muted mx-auto mt-1.5 max-w-md text-xs leading-relaxed">
+            כשיש בתיק פרוטוקול של ועדת-הערר וטיעונים מאוגדים, הניתוח מזהה מה
+            התחזק ומה נטען לראשונה בדיון. ההרצה נשלחת למנתח המשפטי ורצה ברקע.
+          </p>
+          <div className="mt-4 flex justify-center">{runButton}</div>
+        </div>
+      ) : (
+        <>
+          <div className="flex justify-end">{runButton}</div>
+          {/* hearing header strip */}
+          {header && (
+            <div className="bg-parchment border-rule rounded-lg border px-5 py-4 shadow-sm">
+              <div className="flex flex-wrap items-center gap-3">
+                <span className="bg-gold-wash border-rule flex size-9 flex-none items-center justify-center rounded-lg border">
+                  <CalendarDays className="text-gold-deep size-5" aria-hidden />
+                </span>
+                <div className="flex-1">
+                  <div className="text-navy text-[0.95rem] font-bold">
+                    {header.protocol_title || "פרוטוקול דיון"}
+                  </div>
+                  <div className="text-ink-muted mt-0.5 text-xs">
+                    {header.hearing_date
+                      ? `תאריך דיון: ${header.hearing_date}`
+                      : "תאריך דיון לא זוהה"}
+                    {header.panel_members.length > 0 &&
+                      ` · מותב: ${header.panel_members.join(", ")}`}
+                  </div>
+                </div>
+                <Badge
+                  variant="outline"
+                  className="bg-info-bg text-info border-info/30 flex-none rounded-full text-[0.72rem] font-semibold"
+                >
+                  פרוטוקול ועדת-הערר
+                </Badge>
+              </div>
+              {(header.appellants_present.length > 0 ||
+                header.respondents_present.length > 0) && (
+                <div className="border-rule-soft mt-3 flex flex-wrap gap-6 border-t pt-3">
+                  <AttendeeColumn
+                    label="נכחו מטעם העוררים"
+                    dotTone="bg-info"
+                    names={header.appellants_present}
+                  />
+                  <AttendeeColumn
+                    label="נכחו מטעם המשיבים"
+                    dotTone="bg-gold-deep"
+                    names={header.respondents_present}
+                  />
+                </div>
+              )}
+            </div>
+          )}
+
+          {/* filter pills */}
+          <div className="flex flex-wrap items-center gap-2">
+            <FilterPill
+              active={filter === "all"}
+              onClick={() => setFilter("all")}
+              label="הכל"
+              count={data.total}
+            />
+            {strengthened > 0 && (
+              <FilterPill
+                active={filter === "strengthened"}
+                onClick={() => setFilter("strengthened")}
+                label="התחזקו"
+                count={strengthened}
+                icon={<ArrowUp className="size-3.5" aria-hidden />}
+              />
+            )}
+            {newlyRaised > 0 && (
+              <FilterPill
+                active={filter === "newly_raised"}
+                onClick={() => setFilter("newly_raised")}
+                label="נטענו לראשונה"
+                count={newlyRaised}
+                icon={<Plus className="size-3.5" aria-hidden />}
+              />
+            )}
+            {dropped > 0 && (
+              <FilterPill
+                active={filter === "dropped"}
+                onClick={() => setFilter("dropped")}
+                label="נזנחו"
+                count={dropped}
+                icon={<ArrowDown className="size-3.5" aria-hidden />}
+              />
+            )}
+          </div>
+
+          {/* per-party sections */}
+          {sections.map(({ role, rows }) => (
+            <div key={role || "unassigned"} className="space-y-2.5">
+              <div className="flex items-center gap-2.5">
+                <span
+                  className={`h-5 w-1.5 flex-none rounded ${PARTY_MARK_TONE[role]}`}
+                  aria-hidden
+                />
+                <span className="text-navy text-[0.95rem] font-bold">
+                  {PROTOCOL_PARTY_LABELS_HE[role]}
+                </span>
+                <span className="text-ink-muted text-xs">{rows.length}</span>
+              </div>
+              <div className="space-y-2.5">
+                {rows.map((r) => (
+                  <DevelopmentCard key={r.id} row={r} />
+                ))}
+              </div>
+            </div>
+          ))}
+        </>
+      )}
+    </div>
+  );
+}
+
+function FilterPill({
+  active,
+  onClick,
+  label,
+  count,
+  icon,
+}: {
+  active: boolean;
+  onClick: () => void;
+  label: string;
+  count: number;
+  icon?: React.ReactNode;
+}) {
+  return (
+    <button
+      type="button"
+      onClick={onClick}
+      className={`inline-flex items-center gap-1.5 rounded-full border px-3.5 py-1.5 text-xs font-semibold transition-colors ${
+        active
+          ? "bg-navy border-navy text-white"
+          : "bg-surface border-rule text-ink-soft hover:bg-parchment"
+      }`}
+    >
+      {icon}
+      {label}
+      <span className="font-bold">{count}</span>
+    </button>
+  );
+}

web-ui/src/components/cases/positions-panel.tsx

@@ -8,10 +8,6 @@ import { SubsectionCard } from "@/components/compose/subsection-card";
 import { PrecedentsSection } from "@/components/compose/precedents-section";
 import { Markdown } from "@/components/ui/markdown";
 import { useCase } from "@/lib/api/cases";
-import {
-  useCaseLearningStatus,
-  type CaseLearningStatus,
-} from "@/lib/api/learning";
 import { useResearchAnalysis } from "@/lib/api/research";
 import { useCasePrecedents } from "@/lib/api/precedents";
 
@@ -24,45 +20,6 @@ import { useCasePrecedents } from "@/lib/api/precedents";
  * verification moved to their own top-level tabs; /compose was deleted.
  */
 
-// ── Staged-pipeline indicator text — derived from the live learning-status,
-// same source as the drafts-panel LearningStatusBadges. ──────────────────────
-function voiceLearningText(s?: CaseLearningStatus): string {
-  if (!s?.final_uploaded) return "ממתין להעלאת הסופי";
-  const v = s.voice_learning;
-  if (v.outcome === "succeeded") {
-    const bits = [`${v.lessons_count} לקחים הופקו`];
-    if (v.lessons_proposed > 0) bits.push(`${v.lessons_proposed} הוצעו לאישור`);
-    return `✓ הושלם · ${bits.join(" · ")}`;
-  }
-  if (v.outcome === "failed") return v.error ? `✗ נכשל — ${v.error}` : "✗ נכשל";
-  return "ממתין להרצה";
-}
-
-function halachaExtractionText(s?: CaseLearningStatus): string {
-  if (!s?.final_uploaded) return "ממתין להעלאת הסופי";
-  const h = s.halacha_extraction;
-  if (!h.enrolled_in_corpus)
-    return h.not_enrolled_reason ?? "לא נכנס לקורפוס-הפסיקה";
-  switch (h.status) {
-    case "completed":
-      return `✓ הושלם · חולצו ${h.halachot_count} · ${h.approved} אושרו · ${h.rejected} נדחו`;
-    case "processing":
-      return "רץ עכשיו…";
-    case "pending":
-    case "busy":
-      return "בתור";
-    case "partial":
-      return `חלקי · חולצו ${h.halachot_count}`;
-    case "failed":
-    case "extraction_failed":
-      return "✗ נכשל";
-    case "no_chunks":
-      return "אין טקסט לחילוץ";
-    default:
-      return "ממתין להרצה";
-  }
-}
-
 function ProseSection({ title, content }: { title: string; content?: string }) {
   if (!content?.trim()) return null;
   return (
@@ -75,7 +32,9 @@ function ProseSection({ title, content }: { title: string; content?: string }) {
   );
 }
 
-// ── "השלמה והעברה" rail card — DOCX export, analysis upload/download (real) ──
+// ── "ייצוא ועדכון הניתוח" rail card — round-trips the analysis-and-research.md
+// research analysis: export DOCX, upload an updated version, download the raw MD.
+// (Post-final learning status lives on the drafts tab, not here — #226.) ──────
 function FinishRail({
   caseNumber,
   hasAnalysis,
@@ -88,7 +47,6 @@ function FinishRail({
   const fileRef = useRef<HTMLInputElement>(null);
   const [uploading, setUploading] = useState(false);
   const [uploadMsg, setUploadMsg] = useState<{ ok: boolean; text: string } | null>(null);
-  const learning = useCaseLearningStatus(caseNumber);
 
   async function handleUpload(file: File) {
     setUploading(true);
@@ -119,9 +77,12 @@ function FinishRail({
   }
 
   return (
-    <Card className="bg-surface border-rule shadow-sm">
+    <Card className="bg-surface border-rule shadow-sm h-full">
       <CardContent className="px-4 py-4">
-        <h3 className="text-navy text-[0.9rem] font-semibold mb-3">השלמה והעברה</h3>
+        <h3 className="text-navy text-[0.9rem] font-semibold mb-1">ייצוא ועדכון הניתוח</h3>
+        <p className="text-[0.72rem] text-ink-muted mb-3">
+          פעולות על קובץ ניתוח-המחקר (analysis-and-research.md).
+        </p>
 
         <input
           ref={fileRef}
@@ -176,16 +137,6 @@ function FinishRail({
             {uploadMsg.text}
           </p>
         )}
-
-        {/* stage indicators — informational pointers, not actions */}
-        <div className="mt-3 space-y-0">
-          <div className="text-[0.78rem] text-ink-muted pt-2 border-t border-rule-soft">
-            <b className="text-navy">הרץ למידת-קול</b> — {voiceLearningText(learning.data)}
-          </div>
-          <div className="text-[0.78rem] text-ink-muted pt-2 mt-2 border-t border-rule-soft">
-            <b className="text-navy">הרץ אימות-הלכות</b> — {halachaExtractionText(learning.data)}
-          </div>
-        </div>
       </CardContent>
     </Card>
   );
@@ -214,10 +165,6 @@ export function PositionsPanel({ caseNumber }: { caseNumber: string }) {
 
   return (
     <div className="space-y-4">
-      <p className="text-ink-muted text-[0.78rem]">
-        סוגיות-המחלוקת ועמדות הצדדים. עורך עמדת-היו״ר נשמר אוטומטית ומזין את בלוק י׳ (דיון והכרעה).
-      </p>
-
       {analysis.isPending ? (
         <Card className="bg-surface border-rule shadow-sm">
           <CardContent className="px-6 py-5 space-y-3">
@@ -322,9 +269,10 @@ export function PositionsPanel({ caseNumber }: { caseNumber: string }) {
         </div>
       ) : null}
 
-      {/* case-level פסיקה מצורפת + finish/export gates (INV-IA3: not removed) */}
+      {/* case-level פסיקה מצורפת + analysis export/update (INV-IA3: not removed).
-      <div className="grid gap-4 lg:grid-cols-2 items-start pt-2">
+          items-stretch keeps both cards the same height (#226). */}
-        <Card className="bg-surface border-rule shadow-sm">
+      <div className="grid gap-4 lg:grid-cols-2 items-stretch pt-2">
+        <Card className="bg-surface border-rule shadow-sm h-full">
           <CardContent className="px-5 py-4">
             <h3 className="text-navy text-[0.9rem] font-semibold mb-1">פסיקה מצורפת</h3>
             <p className="text-[0.72rem] text-ink-muted mb-3">

web-ui/src/lib/api/generate.ts

@@ -9,7 +9,7 @@
  * Generation runs on the host (claude_session → claude -p), NOT in the container,
  * so the trigger endpoints return 202-accepted and the UI POLLS for completion:
  *   • summary → GET /api/cases/{n}/research/party-claims-summary (200 = ready)
- *   • interim draft → the exports list grows a "טיוטת-ביניים-{n}-vN.docx" file
+ *   • interim draft → the exports list grows a "טיוטה-טענות_הצדדים_{N}.docx" file
  *     (polled via useExports — see exports.ts).
  *
  * The full-decision export ("הפק טיוטת החלטה מלאה") is NOT here — it reuses the
@@ -29,12 +29,58 @@ export type GenerateTriggerResult = {
   error?: string;
 };
 
+/** The two host-side background generation actions (poll target of the status chip). */
+export type GenerateAction = "party_claims_summary" | "interim_draft";
+
 export const generateKeys = {
   all: ["generate"] as const,
   partyClaimsSummary: (caseNumber: string) =>
     [...generateKeys.all, "party-claims-summary", caseNumber] as const,
+  status: (caseNumber: string, action: GenerateAction) =>
+    [...generateKeys.all, "status", action, caseNumber] as const,
 };
 
+/**
+ * Server-derived background-run status (#227 ג). Reconstructed on the backend
+ * from the CEO child issue + its heartbeat_run + the output artifact — so it is
+ * correct after navigating away and back; it never depends on the browser having
+ * stayed on the page. `started_at`/`finished_at` are server ISO-8601 UTC stamps,
+ * so the client renders a correct elapsed time on return.
+ */
+export type GenerateState = "idle" | "queued" | "running" | "done" | "failed";
+
+export type GenerateStatus = {
+  case_number: string;
+  action: GenerateAction;
+  state: GenerateState;
+  started_at: string | null;
+  finished_at: string | null;
+  output_ready: boolean;
+  detail: string;
+  issue_id: string | null;
+};
+
+export function useGenerateStatus(
+  caseNumber: string | undefined,
+  action: GenerateAction,
+) {
+  return useQuery<GenerateStatus>({
+    queryKey: generateKeys.status(caseNumber ?? "", action),
+    queryFn: ({ signal }) =>
+      apiRequest<GenerateStatus>(
+        `/api/cases/${caseNumber}/generate/${action}/status`,
+        { signal },
+      ),
+    enabled: Boolean(caseNumber),
+    // Poll only while a run is in flight; a terminal state stops the interval.
+    refetchInterval: (query) => {
+      const s = query.state.data?.state;
+      return s === "running" || s === "queued" ? 5_000 : false;
+    },
+    staleTime: 2_000,
+  });
+}
+
 /* ── Read: party-claims executive summary (poll target for button 1) ──
    404 (not generated yet) is a normal "not ready" state, not an error — we
    coerce it to null so the panel renders the empty/idle state cleanly. */
@@ -70,23 +116,33 @@ export function useGeneratePartyClaimsSummary(caseNumber: string) {
         { method: "POST" },
       ),
     onSuccess: () => {
-      // Start polling fresh — the file will appear after the host run completes.
+      // Start polling fresh — the file will appear after the host run completes,
+      // and the status chip should pick up the new queued/running run.
       qc.invalidateQueries({
         queryKey: generateKeys.partyClaimsSummary(caseNumber),
       });
+      qc.invalidateQueries({
+        queryKey: generateKeys.status(caseNumber, "party_claims_summary"),
+      });
     },
   });
 }
 
 /* ── Trigger: interim ("party-claims") partial draft (button 2) ── */
 export function useGenerateInterimDraft(caseNumber: string) {
+  const qc = useQueryClient();
   return useMutation({
     mutationFn: () =>
       apiRequest<GenerateTriggerResult>(
         `/api/cases/${caseNumber}/generate/interim-draft`,
         { method: "POST" },
       ),
-    // The result lands in the exports list (טיוטת-ביניים-*.docx); useExports
+    // The result lands in the exports list (טיוטה-טענות_הצדדים_*.docx, polled by
-    // already polls on a 5s interval, so no extra invalidation is needed here.
+    // useExports); re-arm the status chip so it shows the new run immediately.
+    onSuccess: () => {
+      qc.invalidateQueries({
+        queryKey: generateKeys.status(caseNumber, "interim_draft"),
+      });
+    },
   });
 }

web-ui/src/lib/api/protocol-analysis.ts

@@ -0,0 +1,134 @@
+/**
+ * Protocol comparative analysis — the "מה קרה בדיון" panel (#226).
+ *
+ * `analyze_protocol` compares the ערר hearing protocol against the written
+ * pleadings and records, per argument, whether it was strengthened, newly
+ * raised, or dropped at the hearing — plus the sharpened legal question and a
+ * verbatim evidence quote. This module exposes the read + trigger endpoints.
+ */
+
+import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
+import { apiRequest } from "./client";
+import type { LegalArgumentParty } from "./legal-arguments";
+
+export type ProtocolChangeType = "strengthened" | "newly_raised" | "dropped";
+
+/** Empty string = an unassigned party_role on a verdict row. */
+export type ProtocolPartyRole = LegalArgumentParty | "";
+
+export type ProtocolAnalysisRow = {
+  id: string;
+  case_id: string;
+  document_id: string;
+  party_role: ProtocolPartyRole;
+  change_type: ProtocolChangeType;
+  /** The pleaded argument this verdict matched (strengthened/dropped); null for newly_raised. */
+  argument_id: string | null;
+  argument_title: string;
+  summary: string;
+  sharpened_question: string;
+  evidence_quote: string;
+  page_number: number | null;
+  created_at?: string;
+};
+
+export type ProtocolHeader = {
+  hearing_date: string | null;
+  protocol_title: string;
+  protocol_document_id: string;
+  panel_members: string[];
+  appellants_present: string[];
+  respondents_present: string[];
+};
+
+export type ProtocolAnalysisResponse = {
+  case_number: string;
+  total: number;
+  header: ProtocolHeader;
+  by_change: Partial<Record<ProtocolChangeType, number>>;
+  by_party: Partial<Record<ProtocolPartyRole, ProtocolAnalysisRow[]>>;
+  analysis: ProtocolAnalysisRow[];
+};
+
+export const protocolAnalysisKeys = {
+  all: ["protocol-analysis"] as const,
+  byCase: (caseNumber: string) =>
+    [...protocolAnalysisKeys.all, caseNumber] as const,
+};
+
+export function useProtocolAnalysis(caseNumber: string | undefined) {
+  return useQuery({
+    queryKey: protocolAnalysisKeys.byCase(caseNumber ?? ""),
+    queryFn: ({ signal }) =>
+      apiRequest<ProtocolAnalysisResponse>(
+        `/api/cases/${caseNumber}/protocol-analysis`,
+        { signal },
+      ),
+    enabled: Boolean(caseNumber),
+    staleTime: 10_000,
+  });
+}
+
+/**
+ * The analysis runs on the legal-analyst agent (host-side, where the `claude`
+ * CLI lives) — NOT inline in the FastAPI container. The endpoint delegates via
+ * a Paperclip wakeup (`queued`), or reports `skipped` when no analyst route is
+ * available (the chair can then run the MCP tool manually).
+ */
+export type AnalyzeProtocolResult =
+  | {
+      status: "queued";
+      sub_issue_id: string;
+      analyst_id: string;
+      main_issue_id: string;
+    }
+  | {
+      status: "skipped";
+      reason: "no_api_key" | "no_analyst" | "no_issue" | string;
+      company_id?: string;
+    };
+
+export function useAnalyzeProtocol(caseNumber: string | undefined) {
+  const qc = useQueryClient();
+  return useMutation({
+    mutationFn: (documentId?: string) =>
+      apiRequest<AnalyzeProtocolResult>(
+        `/api/cases/${caseNumber}/analyze-protocol${
+          documentId ? `?document_id=${encodeURIComponent(documentId)}` : ""
+        }`,
+        { method: "POST" },
+      ),
+    onSuccess: () => {
+      if (caseNumber) {
+        qc.invalidateQueries({
+          queryKey: protocolAnalysisKeys.byCase(caseNumber),
+        });
+      }
+    },
+  });
+}
+
+export const CHANGE_LABELS_HE: Record<ProtocolChangeType, string> = {
+  strengthened: "התחזק בדיון",
+  newly_raised: "נטען לראשונה",
+  dropped: "נזנח בדיון",
+};
+
+export const PROTOCOL_PARTY_LABELS_HE: Record<ProtocolPartyRole, string> = {
+  appellant: "עוררים",
+  respondent: "משיבים",
+  committee: "ועדה מקומית",
+  permit_applicant: "מבקשי היתר",
+  unknown: "צד לא מזוהה",
+  "": "צד לא מסווג",
+};
+
+/** Display order for the per-party sections. */
+export const PROTOCOL_PARTY_ORDER: ProtocolPartyRole[] = [
+  "appellant",
+  "committee",
+  "respondent",
+  "permit_applicant",
+  "unknown",
+  "",
+];

web/agent_health.py

@@ -0,0 +1,110 @@
+"""Agent health taxonomy (TaskMaster #222) — one crisp state per stuck issue.
+
+Classifies each open, agent-assigned issue into a single health state so the
+stranded-child / recovery-loop cases we diagnose by hand today
+([[reference_recovery_loop_stranded_child]], [[reference_paperclip_recovery_loops]])
+surface automatically in the dashboard — the ``gt feed --problems`` idea from
+Gastown, grounded in *our* failure modes.
+
+Two layers, split for testability (same shape as web.agent_telemetry):
+- :func:`classify_issue_health` is a **pure function** over three primitives
+  (live-run flag + wakeup counts). No DB, no platform symbols — trivially
+  unit-tested and platform-agnostic (INV-G12).
+- The Paperclip-specific fetch that derives those primitives from
+  ``heartbeat_runs`` / ``agent_wakeup_requests`` lives in the shell
+  (``web.paperclip_client.get_agent_health``); this module only decides *how to
+  label* what the fetch measured, so the taxonomy survives a platform swap.
+
+The wakeup/recovery signals it reads are the very ones #219 telemetry now emits
+(``agent.wakeup`` with recovery reasons) — this is the consumer that turns that
+stream into an at-a-glance verdict.
+"""
+from __future__ import annotations
+
+# Severity order (worst first) — the dashboard sorts problems by this.
+HEALTH_STATES: tuple[str, ...] = ("zombie", "stalled", "working", "idle")
+
+# Substrings of Paperclip wakeup ``reason`` values that mark a recovery-loop
+# wake (as opposed to a genuine new task). Sourced from the recovery-loop
+# forensics in memory: an issue that keeps getting these while agent-owned is a
+# zombie — assigned + "active" but nothing is really executing.
+RECOVERY_REASON_MARKERS: tuple[str, ...] = (
+    "source_scoped_recovery",
+    "stranded_assigned",
+    "issue_continuation",
+    "issue_reopened",
+    "missing_disposition",
+    "successful_run_missing_state",
+    "waiting_on_review",
+)
+
+# ≥ this many wakeups inside the fetch window, with no recovery marker and no
+# live run, reads as "poked repeatedly but not progressing" = stalled.
+STALL_WAKEUP_THRESHOLD = 3
+
+
+def is_recovery_reason(reason: str | None) -> bool:
+    """True if a wakeup ``reason`` is a recovery-loop marker (not a new task)."""
+    if not reason:
+        return False
+    return any(marker in reason for marker in RECOVERY_REASON_MARKERS)
+
+
+def classify_issue_health(
+    *,
+    has_live_run: bool,
+    recovery_wakeups: int,
+    total_wakeups: int,
+) -> str:
+    """Label one open, agent-assigned issue. Pure — no I/O.
+
+    - ``working`` — a live run is executing right now.
+    - ``zombie`` — no live run but recovery-loop wakeups are firing: the issue
+      looks agent-owned + active yet nothing is really running (the stranded
+      child we chase manually).
+    - ``stalled`` — no live run, no recovery markers, but poked ``>=``
+      :data:`STALL_WAKEUP_THRESHOLD` times in-window (repeated wakes, no
+      progress).
+    - ``idle`` — open + assigned but quiet: benign waiting.
+
+    Called only for issues that are already open and agent-assigned; unassigned
+    or closed issues are not health-tracked (the fetch filters them out).
+    """
+    if has_live_run:
+        return "working"
+    if recovery_wakeups >= 1:
+        return "zombie"
+    if total_wakeups >= STALL_WAKEUP_THRESHOLD:
+        return "stalled"
+    return "idle"
+
+
+# A zombie needs at least this many recovery-marker wakeups before the watchdog
+# auto-escalates it — filters a single transient recovery wake from a genuine,
+# budget-burning loop. Escalation reassigns the issue to the chair (via
+# escalate_issue), so an escalated issue is agent-unowned next sweep and never
+# re-escalates (idempotent by construction).
+ZOMBIE_ESCALATE_MIN_RECOVERY = 2
+
+
+def zombie_escalation(item: dict) -> tuple[str, str] | None:
+    """Decide whether a health item warrants auto-escalation to the chair.
+
+    Pure (no I/O) so the watchdog's policy is unit-tested. Returns
+    ``(severity, reason)`` for a persistent zombie, else ``None``. Severity
+    scales with loop intensity — a long-running loop (``>=5`` recovery wakes)
+    is ``high``, a fresh one ``medium``. Only ``zombie`` items escalate: a
+    ``stalled``/``working``/``idle`` issue is not a recovery loop.
+    """
+    if item.get("health") != "zombie":
+        return None
+    recovery = int(item.get("recovery_wakeups", 0))
+    if recovery < ZOMBIE_ESCALATE_MIN_RECOVERY:
+        return None
+    severity = "high" if recovery >= 5 else "medium"
+    reason = (
+        f"auto-escalation (watchdog): זוהתה לולאת-recovery — {recovery} יקיצות-שחזור "
+        f"על {item.get('identifier')} ({item.get('agent_name')}); ה-issue נמסר לחיים "
+        f"במקום להמשיך לשרוף budget."
+    )
+    return severity, reason

web/agent_platform_port.py

@@ -33,14 +33,15 @@ from web.paperclip_api import (
     pc_request,
     require_paperclip_db_url,
 )
+from web.agent_telemetry import instrument
 from web.paperclip_client import (
     COMPANIES as PAPERCLIP_COMPANIES,
     accept_interaction as pc_accept_interaction,
     archive_project as pc_archive_project,
-    cancel_interaction as pc_cancel_interaction,
-    cancel_run as pc_cancel_run,
     create_project as pc_create_project,
     create_workflow_issue as pc_create_workflow_issue,
+    get_agent_health as pc_get_agent_health,
+    get_predecessor_context as pc_get_predecessor_context,
     get_agents_for_case as pc_get_agents_for_case,
     get_agents_for_company as pc_get_agents,
     get_case_issues as pc_get_case_issues,
@@ -50,23 +51,82 @@ from web.paperclip_client import (
     get_run_events as pc_get_run_events,
     get_run_log as pc_get_run_log,
     list_live_runs as pc_list_live_runs,
-    post_comment as pc_post_comment,
-    reap_stale_interactions as pc_reap_stale_interactions,
     reject_interaction as pc_reject_interaction,
-    reset_agent_session as pc_reset_agent_session,
-    reset_case_agents as pc_reset_case_agents,
     respond_to_interaction as pc_respond_to_interaction,
     restore_project as pc_restore_project,
     update_project_name as pc_update_project_name,
-    wake_analyst_for_appraiser_facts as pc_wake_analyst_for_appraiser_facts,
+    get_generation_run_status as pc_get_generation_run_status,
-    wake_analyst_for_argument_aggregation as pc_wake_analyst_for_argument_aggregation,
+    # ── raw imports: wrapped with telemetry below (state-affecting ops, #219) ──
-    wake_ceo_agent as pc_wake_ceo,
+    cancel_interaction as _cancel_interaction,
-    wake_ceo_for_action as pc_wake_ceo_for_action,
+    cancel_run as _cancel_run,
-    wake_ceo_for_feedback_fold as pc_wake_ceo_for_feedback_fold,
+    escalate_issue as _escalate_issue,
-    wake_curator_for_final as pc_wake_curator_for_final,
+    post_comment as _post_comment,
-    wake_for_precedent_extraction as pc_wake_for_precedent_extraction,
+    reap_stale_interactions as _reap_stale_interactions,
+    reset_agent_session as _reset_agent_session,
+    reset_case_agents as _reset_case_agents,
+    wake_analyst_for_appraiser_facts as _wake_analyst_for_appraiser_facts,
+    wake_analyst_for_argument_aggregation as _wake_analyst_for_argument_aggregation,
+    wake_analyst_for_protocol_analysis as _wake_analyst_for_protocol_analysis,
+    wake_ceo_agent as _wake_ceo,
+    wake_ceo_for_action as _wake_ceo_for_action,
+    wake_ceo_for_feedback_fold as _wake_ceo_for_feedback_fold,
+    wake_curator_for_final as _wake_curator_for_final,
+    wake_for_precedent_extraction as _wake_for_precedent_extraction,
 )
 
+# ── telemetry-wrapped platform ops (#219 / docs/spec/X15) ───────────────────
+# Every state-affecting platform op the Port exposes emits one structured event
+# through web.agent_telemetry, so recovery-loop behaviour (repeated wakeups,
+# stranded dispositions) is observable in real time rather than reconstructed
+# from the Paperclip DB. The wrappers preserve the public ``pc_*`` names and
+# signatures — call sites in app.py are unchanged. Read-only observability ops
+# (list_live_runs / get_run_log / get_issue_comments …) are intentionally NOT
+# instrumented: they neither wake agents nor change disposition state.
+pc_wake_ceo = instrument("agent.wakeup", role="ceo")(_wake_ceo)
+pc_wake_ceo_for_action = instrument(
+    "agent.wakeup", role="ceo", keys=("case_number", "company_id", "action"),
+)(_wake_ceo_for_action)
+pc_wake_ceo_for_feedback_fold = instrument(
+    "agent.wakeup", role="ceo", keys=("feedback_id", "category", "block_id"),
+)(_wake_ceo_for_feedback_fold)
+pc_wake_curator_for_final = instrument(
+    "agent.wakeup", role="curator", keys=("case_number", "company_id", "task"),
+)(_wake_curator_for_final)
+pc_wake_for_precedent_extraction = instrument(
+    "agent.wakeup", role="ceo", keys=("case_law_id", "citation", "practice_area"),
+)(_wake_for_precedent_extraction)
+pc_wake_analyst_for_appraiser_facts = instrument(
+    "agent.wakeup", role="analyst",
+)(_wake_analyst_for_appraiser_facts)
+pc_wake_analyst_for_argument_aggregation = instrument(
+    "agent.wakeup", role="analyst",
+)(_wake_analyst_for_argument_aggregation)
+pc_wake_analyst_for_protocol_analysis = instrument(
+    "agent.wakeup", role="analyst", keys=("case_number", "company_id", "document_id"),
+)(_wake_analyst_for_protocol_analysis)
+pc_post_comment = instrument(
+    "agent.comment", keys=("issue_id", "company_id"),
+)(_post_comment)
+pc_cancel_interaction = instrument(
+    "interaction.cancelled", keys=("issue_id", "interaction_id"),
+)(_cancel_interaction)
+# First-class escalation-to-chair (#218): the loop-safe alternative to leaving an
+# issue agent-owned+blocked. Emits a severity-tagged event so escalations are
+# countable per case in the same stream as the wakeups they replace.
+pc_escalate_issue = instrument(
+    "agent.escalated", keys=("issue_id", "severity", "company_id", "reason"),
+)(_escalate_issue)
+pc_reap_stale_interactions = instrument(
+    "interaction.reaped", result_keys=("cancelled",),
+)(_reap_stale_interactions)
+pc_cancel_run = instrument("run.cancelled", keys=("run_id",))(_cancel_run)
+pc_reset_agent_session = instrument(
+    "agent.session_reset", keys=("agent_id",),
+)(_reset_agent_session)
+pc_reset_case_agents = instrument(
+    "agent.session_reset", keys=("case_number",),
+)(_reset_case_agents)
+
 # ── domain-named lifecycle aliases (preferred for new call sites) ───────────
 archive_case_project = pc_archive_project
 restore_case_project = pc_restore_project
@@ -106,6 +166,8 @@ __all__ = [
     "pc_wake_for_precedent_extraction",
     "pc_wake_analyst_for_appraiser_facts",
     "pc_wake_analyst_for_argument_aggregation",
+    "pc_wake_analyst_for_protocol_analysis",
+    "pc_get_generation_run_status",
     # comments / interactions
     "pc_post_comment",
     "pc_get_issue_comments",
@@ -115,7 +177,10 @@ __all__ = [
     "pc_respond_to_interaction",
     "pc_cancel_interaction",
     "pc_reap_stale_interactions",
+    "pc_escalate_issue",
     # agent-run observability + control (live view + smart management)
+    "pc_get_agent_health",
+    "pc_get_predecessor_context",
     "pc_list_live_runs",
     "pc_get_run_log",
     "pc_get_run_events",

web/agent_telemetry.py

@@ -0,0 +1,157 @@
+"""Agent-platform telemetry (TaskMaster #219) — structured, platform-agnostic events.
+
+Emits one structured event per agent-platform operation (wakeup, disposition
+comment, interaction reap/cancel, run cancel, session reset) so recovery-loop
+behaviour is **observable in real time** instead of reconstructed forensically
+from the Paperclip DB (``agent_wakeup_requests`` / ``heartbeat_runs``). See
+[[reference_paperclip_recovery_loops]] / [[reference_recovery_loop_stranded_child]]
+for the failure modes this makes visible.
+
+**Platform-agnostic by design (INV-G12 / docs/spec/X15).** Events are named in
+domain terms and carry no Paperclip-specific symbols — this module imports
+*nothing* from the Paperclip shell. The Port (``web/agent_platform_port.py``)
+calls :func:`instrument` / :func:`emit` around its operations, so the telemetry
+semantics survive a platform swap.
+
+**OTLP-ready, "logs first" (#219).** Each event is a flat dict with a stable
+schema, written today as one JSON line on the ``agent.telemetry`` logger
+(greppable in Coolify logs). Swapping :data:`_SINK` for an OTLP exporter — or
+attaching an OTLP handler to that logger — is the later step, with **no
+call-site changes**. Timestamps are stored in UTC (display-in-Israel is a
+read-side concern; see [[feedback_israel_time]]).
+
+Single path (INV-G2): the only telemetry emitter for the agent platform. Do not
+add a parallel structured-log format for these events elsewhere — route through
+:func:`emit`.
+"""
+from __future__ import annotations
+
+import functools
+import inspect
+import json
+import logging
+import time
+from collections.abc import Awaitable, Callable, Sequence
+from datetime import datetime, timezone
+from typing import Any, TypeVar
+
+logger = logging.getLogger("agent.telemetry")
+
+# Schema version — bump when the event shape changes so downstream consumers
+# (dashboards, the #222 health taxonomy) can migrate deliberately.
+SCHEMA_VERSION = 1
+
+# Standard argument keys lifted onto every event when present on the wrapped call.
+_STANDARD_KEYS: tuple[str, ...] = ("case_number", "issue_id", "company_id", "reason")
+
+T = TypeVar("T")
+
+
+def _now_iso() -> str:
+    """UTC ISO-8601 with a trailing ``Z`` — store-UTC (feedback_israel_time)."""
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def _sink(payload: dict[str, Any]) -> None:
+    """Where an event goes. Today: a JSON line on the ``agent.telemetry`` logger.
+
+    Isolated so a future OTLP exporter is a one-line swap with no call-site
+    churn (#219 "backend later").
+    """
+    logger.info(json.dumps(payload, ensure_ascii=False, default=str))
+
+
+_SINK: Callable[[dict[str, Any]], None] = _sink
+
+
+def emit(event: str, *, outcome: str = "ok", **fields: Any) -> None:
+    """Emit one structured agent-platform event. Never raises.
+
+    Telemetry must never break the operation it observes — any failure in the
+    sink is swallowed to a warning. ``event`` is a dotted domain name
+    (``agent.wakeup``, ``interaction.reaped``, ``run.cancelled`` …). Extra
+    ``fields`` are merged flat into the event; ``None`` values are dropped to
+    keep events terse.
+    """
+    payload: dict[str, Any] = {
+        "ts": _now_iso(),
+        "v": SCHEMA_VERSION,
+        "event": event,
+        "outcome": outcome,
+    }
+    for key, value in fields.items():
+        if value is not None:
+            payload[key] = value
+    try:
+        _SINK(payload)
+    except Exception:  # pragma: no cover — telemetry is best-effort, never fatal
+        logger.warning("agent.telemetry emit failed for event=%s", event, exc_info=True)
+
+
+def _outcome_of(result: Any) -> str:
+    """Derive an outcome from a platform call's return value.
+
+    Our platform helpers return either a plain dict (Paperclip API JSON) or a
+    ``{"ok": bool, ...}`` envelope. A falsy ``ok`` is a *no-op* (e.g. an
+    interaction that was already resolved), distinct from a raised error.
+    """
+    if isinstance(result, dict) and result.get("ok") is False:
+        return "noop"
+    return "ok"
+
+
+def _duration_ms(t0: float) -> int:
+    return int((time.monotonic() - t0) * 1000)
+
+
+def instrument(
+    event: str,
+    *,
+    role: str | None = None,
+    keys: Sequence[str] = _STANDARD_KEYS,
+    result_keys: Sequence[str] = (),
+    **static: Any,
+) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
+    """Decorate an async platform op so it emits a structured event per call.
+
+    Fields are pulled from the call's bound arguments (``keys`` present on the
+    signature), plus a static ``role`` and any ``static`` attributes, plus
+    ``duration_ms`` and an ``outcome`` derived from the result (or ``error`` on
+    exception). ``result_keys`` lifts named keys out of a dict return value onto
+    the event (e.g. the ``cancelled`` count from the reaper).
+
+    The emit is wrapped so a telemetry bug can never fail the underlying call.
+    """
+
+    def deco(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
+        sig = inspect.signature(fn)
+
+        @functools.wraps(fn)
+        async def wrapper(*args: Any, **kwargs: Any) -> T:
+            try:
+                bound = sig.bind(*args, **kwargs)
+                bound.apply_defaults()
+                arguments = bound.arguments
+            except TypeError:
+                arguments = {}
+            fields: dict[str, Any] = {k: arguments.get(k) for k in keys}
+            if role is not None:
+                fields["agent_role"] = role
+            fields.update(static)
+
+            t0 = time.monotonic()
+            try:
+                result = await fn(*args, **kwargs)
+            except Exception as exc:
+                emit(event, outcome="error", error=repr(exc), duration_ms=_duration_ms(t0), **fields)
+                raise
+            if isinstance(result, dict):
+                for rk in result_keys:
+                    if rk in result:
+                        fields[rk] = result[rk]
+            emit(event, outcome=_outcome_of(result), duration_ms=_duration_ms(t0), **fields)
+            return result
+
+        return wrapper
+
+    return deco

web/app.py

@@ -61,6 +61,9 @@ from web.agent_platform_port import (
     pc_cancel_run,
     pc_create_project,
     pc_create_workflow_issue,
+    pc_escalate_issue,
+    pc_get_agent_health,
+    pc_get_predecessor_context,
     pc_get_agents,
     pc_get_agents_for_case,
     pc_get_case_issues,
@@ -79,6 +82,8 @@ from web.agent_platform_port import (
     pc_restore_project,
     pc_wake_analyst_for_appraiser_facts,
     pc_wake_analyst_for_argument_aggregation,
+    pc_wake_analyst_for_protocol_analysis,
+    pc_get_generation_run_status,
     rename_case_project,
     pc_wake_ceo,
     pc_wake_ceo_for_action,
@@ -119,18 +124,66 @@ async def _interaction_reaper_loop():
             logger.warning("interaction reaper sweep failed", exc_info=True)
 
 
+# Watchdog kill-switch — set AGENT_ZOMBIE_AUTO_ESCALATE=0 to observe health
+# without acting (escalation stays available manually via the endpoint).
+_ZOMBIE_AUTO_ESCALATE = os.environ.get("AGENT_ZOMBIE_AUTO_ESCALATE", "1") != "0"
+
+
+async def _zombie_escalation_loop():
+    """Watchdog: auto-escalate persistent recovery-loop zombies to the chair (#218+#222).
+
+    Every 15 min, derive per-issue health (#222) and hand any persistent zombie
+    (a recovery loop, ``>=`` ``ZOMBIE_ESCALATE_MIN_RECOVERY`` recovery wakeups)
+    to the human via the loop-safe escalate primitive (#218) — closing the loop
+    that today burns budget until a human notices. Server-side on purpose: an
+    agent stuck in a recovery loop cannot reliably self-escalate. Idempotent —
+    an escalated issue is agent-unowned next sweep, so it is not re-escalated.
+    """
+    from web.agent_health import zombie_escalation
+
+    while True:
+        try:
+            await asyncio.sleep(900)
+            if not _ZOMBIE_AUTO_ESCALATE:
+                continue
+            health = await pc_get_agent_health()
+            for item in health.get("items", []):
+                decision = zombie_escalation(item)
+                if decision is None:
+                    continue
+                severity, reason = decision
+                result = await pc_escalate_issue(item["issue_id"], severity, reason)
+                if result.get("ok"):
+                    logger.warning(
+                        "watchdog auto-escalated %s (%s, %d recovery wakeups) to chair [%s]",
+                        item.get("identifier"), item.get("agent_name"),
+                        item.get("recovery_wakeups", 0), severity,
+                    )
+                else:
+                    logger.warning(
+                        "watchdog escalation failed for %s: %s",
+                        item.get("identifier"), result.get("error"),
+                    )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.warning("zombie escalation sweep failed", exc_info=True)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
     await db.init_schema()
     sync_task = asyncio.create_task(git_sync.sweep_loop())
     reaper_task = asyncio.create_task(_interaction_reaper_loop())
+    watchdog_task = asyncio.create_task(_zombie_escalation_loop())
     try:
         yield
     finally:
         sync_task.cancel()
         reaper_task.cancel()
-        for _t in (sync_task, reaper_task):
+        watchdog_task.cancel()
+        for _t in (sync_task, reaper_task, watchdog_task):
             try:
                 await _t
             except asyncio.CancelledError:
@@ -2659,6 +2712,97 @@ async def api_get_legal_arguments(case_number: str, party: str = ""):
     }
 
 
+@app.get("/api/cases/{case_number}/protocol-analysis")
+async def api_get_protocol_analysis(case_number: str):
+    """Hearing-protocol comparative analysis for the "מה קרה בדיון" panel (#226).
+
+    Read-only surface over the case-knowledge produced by ``analyze_protocol``:
+    the hearing header (date + who appeared, from the canonical case columns)
+    plus the per-argument verdicts (strengthened / newly_raised / dropped)
+    grouped by party. Container-safe — pure DB reads, no LLM. Returns an empty
+    analysis (not an error) when the analysis has not been run yet.
+    """
+    case = await db.get_case_by_number(case_number)
+    if not case:
+        raise HTTPException(404, f"תיק {case_number} לא נמצא")
+
+    case_id = UUID(case["id"])
+    rows = await db.list_protocol_analysis(case_id)
+
+    # Protocol title comes from the analysed document (all rows share one
+    # document_id per idempotent replace). Resolve it via the case's documents.
+    protocol_title = ""
+    protocol_document_id = ""
+    if rows:
+        protocol_document_id = rows[0].get("document_id") or ""
+        docs = await db.list_documents(case_id)
+        match = next(
+            (d for d in docs if str(d.get("id")) == protocol_document_id), None,
+        )
+        if match:
+            protocol_title = match.get("title") or ""
+
+    attendees = case.get("hearing_attendees") or {}
+    hearing_date = case.get("hearing_date")
+    header = {
+        "hearing_date": hearing_date.isoformat() if hearing_date else None,
+        "protocol_title": protocol_title,
+        "protocol_document_id": protocol_document_id,
+        "panel_members": attendees.get("panel_members") or [],
+        "appellants_present": attendees.get("appellants_present") or [],
+        "respondents_present": attendees.get("respondents_present") or [],
+    }
+
+    # Group verdicts by party for the panel's per-side sections, in display order.
+    by_party: dict[str, list[dict]] = {}
+    by_change: dict[str, int] = {}
+    for r in rows:
+        by_party.setdefault(r.get("party_role") or "", []).append(r)
+        ct = r.get("change_type", "")
+        by_change[ct] = by_change.get(ct, 0) + 1
+
+    return {
+        "case_number": case_number,
+        "total": len(rows),
+        "header": header,
+        "by_change": by_change,
+        "by_party": by_party,
+        "analysis": rows,
+    }
+
+
+@app.post("/api/cases/{case_number}/analyze-protocol")
+async def api_analyze_protocol(case_number: str, document_id: str = ""):
+    """Queue hearing-protocol analysis by waking the legal-analyst agent (#226).
+
+    Same delegation rationale as ``aggregate-arguments``: ``analyze_protocol``
+    calls the local ``claude`` CLI, absent in this container, so we route to the
+    company's analyst rather than running a doomed in-container BackgroundTask.
+
+    Response: {"status": "queued", ...} or {"status": "skipped", "reason": ...}.
+    """
+    case = await db.get_case_by_number(case_number)
+    if not case:
+        raise HTTPException(404, f"תיק {case_number} לא נמצא")
+
+    prefix = case_number[:1]
+    company_id = (
+        PAPERCLIP_COMPANIES["licensing"] if prefix == "1"
+        else PAPERCLIP_COMPANIES["betterment"] if prefix in ("8", "9")
+        else ""
+    )
+
+    try:
+        result = await pc_wake_analyst_for_protocol_analysis(
+            case_number, company_id=company_id, document_id=document_id,
+        )
+    except Exception as e:
+        logger.exception("analyst wakeup failed for protocol analysis %s", case_number)
+        raise HTTPException(500, f"לא ניתן לשלוח לאנליטיקאי: {e}")
+
+    return result
+
+
 @app.post("/api/cases/{case_number}/direction")
 async def api_set_direction(case_number: str, req: DirectionRequest):
     """Save the approved direction document for the discussion block."""
@@ -3120,7 +3264,7 @@ async def api_party_claims_summary(case_number: str):
 # Wakeup goes through the Paperclip API helper (pc_wake_ceo_for_action → the
 # platform Port → POST /api/agents/{id}/wakeup), NEVER a direct DB insert.
 # Polling for completion uses the existing read endpoints (research/party-claims-
-# summary for the summary; the exports list for the טיוטת-ביניים-*.docx file).
+# summary for the summary; the exports list for the טיוטה-טענות_הצדדים_*.docx file).
 
 
 async def _wake_ceo_action(case_number: str, action: str) -> dict:
@@ -3153,10 +3297,99 @@ async def api_generate_party_claims_summary(case_number: str):
 async def api_generate_interim_draft(case_number: str):
     """Fire-and-accept: wake the CEO to generate the partial "party-claims" draft
     (שלב H → write_interim_draft + export_interim_draft). Runs host-side; poll the
-    exports list for the new טיוטת-ביניים-{case}-vN.docx."""
+    exports list for the new טיוטה-טענות_הצדדים_{N}.docx."""
     return await _wake_ceo_action(case_number, "interim_draft")
 
 
+_GENERATE_ACTIONS = {"party_claims_summary", "interim_draft"}
+# Run statuses (Paperclip heartbeat_runs.status) grouped for state mapping.
+_RUN_ACTIVE = {"running", "in_progress", "queued", "pending", "scheduled", "created"}
+_RUN_FAILED = {"cancelled", "failed", "error", "timed_out", "lost"}
+
+
+def _generation_output_ready(case_number: str, action: str) -> bool:
+    """Whether the generation artifact already exists on disk (host-side)."""
+    if action == "party_claims_summary":
+        from legal_mcp.services import party_claims_summary as pcs
+        return pcs.summary_file_path(case_number).exists()
+    # interim_draft → a "טיוטה-טענות_הצדדים_*.docx" file in the case exports dir.
+    export_dir = config.find_case_dir(case_number) / "exports"
+    if not export_dir.exists():
+        return False
+    return any(
+        f.is_file() and f.suffix.lower() == ".docx" and f.name.startswith("טיוטה-טענות_הצדדים_")
+        for f in export_dir.iterdir()
+    )
+
+
+@app.get("/api/cases/{case_number}/generate/{action}/status")
+async def api_generate_status(case_number: str, action: str):
+    """Server-derived status of a "הפקת מסמכים" background generation (#227 ג).
+
+    Reconstructs {idle|queued|running|done|failed} from the CEO child issue + its
+    latest heartbeat_run (via the platform port) plus whether the output artifact
+    exists — so the UI indicator is correct after navigating away and back; it
+    never depends on the browser staying on the page. ``started_at``/``finished_at``
+    are server timestamps (ISO-8601 UTC) so the client renders a correct elapsed
+    time on return.
+    """
+    if action not in _GENERATE_ACTIONS:
+        raise HTTPException(400, f"action לא תקין: {action}")
+    case = await db.get_case_by_number(case_number)
+    if not case:
+        raise HTTPException(404, f"תיק {case_number} לא נמצא")
+
+    output_ready = _generation_output_ready(case_number, action)
+    try:
+        run = await pc_get_generation_run_status(case_number, action)
+    except Exception as e:
+        logger.warning("generation status lookup failed for %s/%s: %s", case_number, action, e)
+        run = {"issue_id": None}
+
+    run_status = (run.get("run_status") or "").lower()
+    started_at = run.get("started_at")
+    active = run_status in _RUN_ACTIVE or (
+        run.get("issue_status") == "in_progress"
+        and run_status not in _RUN_FAILED
+        and run_status not in {"completed", "succeeded", "done"}
+        and not run.get("finished_at")
+    )
+
+    # Precedence: an in-flight run wins; then a produced artifact; then a failed
+    # run; else idle. (A produced file reads as "done" even if a later retry
+    # failed — the artifact exists.)
+    if active:
+        state = "running" if started_at else "queued"
+    elif output_ready:
+        state = "done"
+    elif run_status in _RUN_FAILED:
+        state = "failed"
+    elif run_status in {"completed", "succeeded", "done"}:
+        # Ran to completion but produced no artifact → treat as a failure so the
+        # chair isn't left thinking it worked.
+        state = "failed"
+    else:
+        state = "idle"
+
+    detail = ""
+    if state == "failed":
+        if run.get("error_code") == "issue_assignee_changed":
+            detail = "הריצה בוטלה (הוקצתה מחדש)"
+        else:
+            detail = run.get("error") or "הריצה נכשלה"
+
+    return {
+        "case_number": case_number,
+        "action": action,
+        "state": state,
+        "started_at": started_at,
+        "finished_at": run.get("finished_at"),
+        "output_ready": output_ready,
+        "detail": detail,
+        "issue_id": run.get("issue_id"),
+    }
+
+
 @app.get("/api/cases/{case_number}/research/party-claims-summary/download")
 async def api_party_claims_summary_download(case_number: str):
     """Download the raw party-claims-summary.md file."""
@@ -4469,6 +4702,30 @@ async def api_reset_case_agents(case_number: str):
     return result
 
 
+class EscalateRequest(BaseModel):
+    issue_id: str
+    severity: Literal["critical", "high", "medium"]
+    reason: str
+
+
+@app.post("/api/cases/{case_number}/agents/escalate")
+async def api_escalate_issue(case_number: str, req: EscalateRequest):
+    """Escalate a single stuck issue to the chair (#218).
+
+    The loop-safe alternative to leaving an issue agent-owned+blocked: hands the
+    issue to the human in one atomic transition (in_review + assignee_user_id)
+    and records a severity note, without re-waking any agent. Emits an
+    ``agent.escalated`` telemetry event via the Port.
+    """
+    issues = await pc_get_case_issues(case_number)
+    if not any(i["id"] == req.issue_id for i in issues):
+        raise HTTPException(404, f"Issue {req.issue_id} לא שייך לתיק {case_number}")
+    result = await pc_escalate_issue(req.issue_id, req.severity, req.reason)
+    if not result.get("ok"):
+        raise HTTPException(400, result.get("error", "ההסלמה נכשלה"))
+    return result
+
+
 # ── Settings: MCP Server Configuration ────────────────────────────
 #
 # Source of truth for legal-ai env vars is Coolify (see memory:
@@ -7324,6 +7581,30 @@ async def operations_agents():
     }
 
 
+@app.get("/api/operations/agents/health")
+async def operations_agent_health():
+    """Per-issue agent health taxonomy (#222): zombie / stalled / working / idle.
+
+    Read-only. Surfaces the stranded-child / recovery-loop cases (``zombie``)
+    automatically — the ``gt feed --problems`` idea grounded in our failure
+    modes — so they no longer need hand-querying the Paperclip DB. Consumed by
+    the /operations dashboard (UI is design-gated, follow-up).
+    """
+    return await pc_get_agent_health()
+
+
+@app.get("/api/operations/issues/{issue_id}/predecessor")
+async def operations_issue_predecessor(issue_id: str, limit: int = 3):
+    """Recent finished runs on an issue for session continuation (#220, "seance").
+
+    Read-only. Lets a fresh heartbeat read what its predecessors on the same
+    issue concluded (the run ``summary`` each leaves) instead of rediscovering
+    context from scratch. Agent-facing wiring (MCP tool + HEARTBEAT) is a
+    follow-up; this is the backend it will call.
+    """
+    return await pc_get_predecessor_context(issue_id, limit)
+
+
 @app.get("/api/operations/agents/runs/{run_id}/log")
 async def operations_agent_run_log(run_id: str):
     """Full output log (NDJSON stream) of one heartbeat run."""

web/paperclip_client.py

@@ -856,6 +856,78 @@ async def reset_case_agents(case_number: str) -> dict:
     }
 
 
+ESCALATION_SEVERITIES = ("critical", "high", "medium")
+
+
+async def escalate_issue(
+    issue_id: str, severity: str, reason: str, company_id: str = "",
+) -> dict:
+    """First-class, severity-routed escalation of a stuck issue to the chair (#218).
+
+    Replaces the fragile manual "PATCH dance" (reassign + set in_review + open an
+    interaction) that repeatedly tripped Paperclip's recovery loops
+    (``source_scoped_recovery_action`` / ``stranded_assigned_issue`` /
+    ``issue_reopened_via_comment``; see memory reference_paperclip_recovery_loops).
+
+    Two effects in **one DB transaction**, mirroring the proven loop-safe path of
+    :func:`reset_case_agents` (raw SQL, not REST):
+
+    1. **Atomic human-owned transition** — a single ``UPDATE`` to the stable end
+       state ``{status:'in_review', assignee_agent_id:null,
+       assignee_user_id:CHAIM_USER_ID}`` (recovery-loops rule #7). Direct-DB on
+       purpose: it bypasses Paperclip's disposition resolver — the very machinery
+       whose multi-PATCH/`issue.released` behaviour *causes* the loops — so the
+       issue lands human-owned in one shot with no ``done→todo`` flip.
+    2. **Durable severity note** — an ``author_type='system'`` comment recording
+       severity + reason. System-authored on purpose: the ``route-pending-comments``
+       sweep routes only ``author_type='user'`` (chair) comments to the CEO, so a
+       system note is inert w.r.t. routing and will **not** re-wake an agent
+       (project_comment_delivery_guarantee). No wakeup is issued — the whole point
+       is to hand off to the human, not re-invoke an agent.
+
+    The ``agent.escalated`` telemetry event is emitted by the Port wrapper
+    (docs/spec/X15) — this is the loop-safe counterpart the CEO/analysts reach
+    for instead of leaving an issue agent-owned+blocked.
+    """
+    if severity not in ESCALATION_SEVERITIES:
+        return {
+            "ok": False,
+            "error": f"invalid severity {severity!r}; expected one of {ESCALATION_SEVERITIES}",
+        }
+
+    body = f"🚨 הסלמה [{severity}] לחיים\n\n{reason}"
+    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+    try:
+        async with conn.transaction():
+            row = await conn.fetchrow(
+                """UPDATE issues
+                      SET status='in_review', assignee_agent_id=null,
+                          assignee_user_id=$1, updated_at=now()
+                    WHERE id=$2::uuid
+                RETURNING id, identifier, company_id""",
+                CHAIM_USER_ID, issue_id,
+            )
+            if not row:
+                return {"ok": False, "error": f"issue {issue_id} not found"}
+            cid = company_id or str(row["company_id"])
+            await conn.execute(
+                """INSERT INTO issue_comments (id, company_id, issue_id, body, author_type)
+                   VALUES ($1::uuid, $2::uuid, $3::uuid, $4, 'system')""",
+                str(uuid.uuid4()), cid, issue_id, body,
+            )
+    finally:
+        await conn.close()
+
+    logger.info("Escalated issue %s to chair [severity=%s]", issue_id, severity)
+    return {
+        "ok": True,
+        "id": str(row["id"]),
+        "identifier": row["identifier"],
+        "severity": severity,
+        "status": "in_review",
+    }
+
+
 async def respond_to_interaction(
     issue_id: str, interaction_id: str, payload: dict,
 ) -> dict:
@@ -962,6 +1034,152 @@ async def reap_stale_interactions() -> dict:
     return {"ok": True, "cancelled": len(rows)}
 
 
+# Windows for the agent-health taxonomy (#222). A run with no finished_at older
+# than the live window is treated as dead (not "working"); wakeups are counted
+# over the recovery window to spot loops.
+_HEALTH_LIVE_RUN_WINDOW = "30 minutes"
+_HEALTH_WAKEUP_WINDOW = "2 hours"
+
+
+async def get_agent_health() -> dict:
+    """Classify every open, agent-assigned issue into a health state (#222).
+
+    Read-only. Derives, per issue, the three primitives the pure classifier
+    (:func:`web.agent_health.classify_issue_health`) needs, from Paperclip's
+    ``heartbeat_runs`` (a live run = ``finished_at IS NULL`` within the live
+    window) and ``agent_wakeup_requests`` (total + recovery-marker counts within
+    the recovery window). Scoped to the two legal-ai companies.
+
+    Surfaces the stranded-child / recovery-loop cases (``zombie``) automatically
+    instead of by hand-querying the DB. Returns items sorted worst-first.
+    """
+    from web.agent_health import (  # local import: keep the shell → agnostic dep inward
+        HEALTH_STATES,
+        classify_issue_health,
+        is_recovery_reason,
+    )
+
+    company_ids = list(COMPANIES.values())
+    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+    try:
+        issues = await conn.fetch(
+            """SELECT i.id, i.identifier, i.status, i.assignee_agent_id,
+                      a.name AS agent_name
+                 FROM issues i
+                 JOIN agents a ON a.id = i.assignee_agent_id
+                WHERE i.company_id = ANY($1::uuid[])
+                  AND i.assignee_agent_id IS NOT NULL
+                  AND i.status IN ('backlog','todo','in_progress','blocked','in_review')""",
+            company_ids,
+        )
+        if not issues:
+            return {"ok": True, "items": [], "counts": {s: 0 for s in HEALTH_STATES}}
+
+        agent_ids = list({str(r["assignee_agent_id"]) for r in issues})
+
+        live_rows = await conn.fetch(
+            f"""SELECT DISTINCT agent_id FROM heartbeat_runs
+                 WHERE agent_id = ANY($1::uuid[])
+                   AND finished_at IS NULL
+                   AND started_at > now() - interval '{_HEALTH_LIVE_RUN_WINDOW}'""",
+            agent_ids,
+        )
+        live_agents = {str(r["agent_id"]) for r in live_rows}
+
+        wake_rows = await conn.fetch(
+            f"""SELECT agent_id, payload->>'issueId' AS issue_id, reason
+                  FROM agent_wakeup_requests
+                 WHERE agent_id = ANY($1::uuid[])
+                   AND requested_at > now() - interval '{_HEALTH_WAKEUP_WINDOW}'""",
+            agent_ids,
+        )
+    finally:
+        await conn.close()
+
+    total_by_issue: dict[str, int] = {}
+    recovery_by_issue: dict[str, int] = {}
+    for w in wake_rows:
+        iid = w["issue_id"]
+        if not iid:
+            continue
+        total_by_issue[iid] = total_by_issue.get(iid, 0) + 1
+        if is_recovery_reason(w["reason"]):
+            recovery_by_issue[iid] = recovery_by_issue.get(iid, 0) + 1
+
+    items = []
+    counts = {s: 0 for s in HEALTH_STATES}
+    for r in issues:
+        iid = str(r["id"])
+        state = classify_issue_health(
+            has_live_run=str(r["assignee_agent_id"]) in live_agents,
+            recovery_wakeups=recovery_by_issue.get(iid, 0),
+            total_wakeups=total_by_issue.get(iid, 0),
+        )
+        counts[state] += 1
+        items.append({
+            "issue_id": iid,
+            "identifier": r["identifier"],
+            "status": r["status"],
+            "agent_id": str(r["assignee_agent_id"]),
+            "agent_name": r["agent_name"],
+            "health": state,
+            "wakeups": total_by_issue.get(iid, 0),
+            "recovery_wakeups": recovery_by_issue.get(iid, 0),
+        })
+
+    order = {s: i for i, s in enumerate(HEALTH_STATES)}
+    items.sort(key=lambda it: (order[it["health"]], -it["recovery_wakeups"]))
+    return {"ok": True, "items": items, "counts": counts}
+
+
+async def get_predecessor_context(issue_id: str, limit: int = 3) -> dict:
+    """Recent finished runs on an issue, newest-first, for session continuation (#220).
+
+    The "seance" idea from Gastown, grounded in our data: each ephemeral heartbeat
+    rediscovers context every wake (the generic *blind heartbeat*; see
+    [[project_comment_delivery_guarantee]]). This lets a fresh wake read what its
+    predecessors on the SAME issue concluded — the run ``summary`` each heartbeat
+    leaves in ``heartbeat_runs.result_json`` — instead of re-deriving from scratch.
+
+    Read-only. Runs are tied to the issue via ``wakeup_request_id`` →
+    ``agent_wakeup_requests.payload->>'issueId'`` (only finished runs are useful
+    as predecessors). ``limit`` is clamped to 1..10.
+    """
+    limit = max(1, min(int(limit), 10))
+    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+    try:
+        rows = await conn.fetch(
+            """SELECT h.id, h.status, h.started_at, h.finished_at,
+                      h.result_json->>'summary' AS summary,
+                      h.error_code, h.session_id_after, a.name AS agent_name
+                 FROM heartbeat_runs h
+                 JOIN agent_wakeup_requests w ON w.id = h.wakeup_request_id
+                 LEFT JOIN agents a ON a.id = h.agent_id
+                WHERE w.payload->>'issueId' = $1
+                  AND h.finished_at IS NOT NULL
+                ORDER BY h.started_at DESC
+                LIMIT $2""",
+            issue_id, limit,
+        )
+    finally:
+        await conn.close()
+
+    runs = [
+        {
+            "run_id": str(r["id"]),
+            "status": r["status"],
+            "started_at": r["started_at"].isoformat() if r["started_at"] else None,
+            "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
+            "summary": r["summary"],
+            "error_code": r["error_code"],
+            "session_id": r["session_id_after"],
+            "agent_name": r["agent_name"],
+        }
+        for r in rows
+    ]
+    return {"ok": True, "issue_id": issue_id, "runs": runs}
+
+
 # Singleton project for the precedent-library extraction queue. One issue per
 # uploaded precedent — assigned to the CEO who runs the local-MCP extractor.
 _LIBRARY_PROJECT_NAME = "ספריית פסיקה — תור חילוץ"
@@ -1258,25 +1476,58 @@ async def wake_ceo_agent(issue_id: str, case_number: str, company_id: str = "")
     return result
 
 
+def _ceo_action_brief(action: str, case_number: str) -> tuple[str, str]:
+    """(child-issue title, description) for a deterministic CEO generation action.
+
+    The instruction is self-contained in the description (like
+    ``wake_analyst_for_argument_aggregation``) so the run does not depend on the
+    CEO parsing ``payload.action`` — it just runs the named tool and closes the
+    child. Both actions run host-side (claude_session → local claude CLI).
+    """
+    if action == "party_claims_summary":
+        title = f"[ערר {case_number}] הפקת סיכום-מנהלים לטענות הצדדים"
+        description = (
+            f"חיים ביקש הפקת סיכום-מנהלים לטענות הצדדים בתיק {case_number}.\n\n"
+            f"הרץ `mcp__legal-ai__summarize_party_claims(case_number=\"{case_number}\")`, "
+            f"כתוב comment קצר בעברית עם תמצית התוצאה, וסגור issue זה כ-done. "
+            f"אם חסרים נתונים (אין טענות מחולצות) — דווח ב-comment מה חסר וסגור כ-blocked."
+        )
+        return title, description
+    # default: interim_draft
+    title = f"[ערר {case_number}] הפקת טיוטת טענות-הצדדים (טיוטת-ביניים)"
+    description = (
+        f"חיים ביקש הפקת טיוטת טענות-הצדדים (טיוטת-ביניים) בתיק {case_number}.\n\n"
+        f"הרץ `mcp__legal-ai__write_interim_draft(case_number=\"{case_number}\")` ואז "
+        f"`mcp__legal-ai__export_interim_draft(case_number=\"{case_number}\")`, כתוב comment "
+        f"קצר בעברית עם שם הקובץ שנוצר, וסגור issue זה כ-done. אם חסרים נתונים — דווח "
+        f"ב-comment וסגור כ-blocked."
+    )
+    return title, description
+
+
 async def wake_ceo_for_action(
     case_number: str,
     action: str,
     company_id: str = "",
 ) -> dict:
-    """Wake the CEO with a deterministic *structured action* on the case's MAIN issue.
+    """Wake the CEO to run a deterministic generation action (TaskMaster #214).
 
-    Drives the UI "הפקת מסמכים" buttons (TaskMaster #214): the CEO reads
+    Drives the UI "הפקת מסמכים" buttons:
-    ``payload.action`` and routes deterministically to its שלב H / שלב H2
+      action="party_claims_summary" → summarize_party_claims (שלב H2)
-    side-quests (legal-ceo.md §"פעולות סטרוקטורליות") — no free-text parsing.
+      action="interim_draft"        → write_interim_draft + export_interim_draft (שלב H)
 
-      action="party_claims_summary" → שלב H2 (summarize_party_claims)
+    A **child issue assigned to the CEO** is created under the case's main issue
-      action="interim_draft"        → שלב H  (write_interim_draft + export_interim_draft)
+    and the wakeup targets that child — NOT the main issue directly (#227). The
+    main issue is usually ``in_review`` assigned to the chair (human) while a case
+    waits; waking the CEO *on* a human-owned issue makes Paperclip cancel the run
+    immediately (``issue_assignee_changed``) — the exact silent failure this
+    fixes. The CEO owns the child, so its run is never cancelled. Same delegation
+    shape as ``wake_analyst_for_argument_aggregation``.
 
-    These are side-quests on the EXISTING ``[ערר {case_number}]`` issue — no child
+    Wakeup goes through the Paperclip API (``POST /api/agents/{id}/wakeup``), never
-    issue is created (the CEO must not change ``cases.status`` or spawn sub-agents).
+    a direct DB insert (CLAUDE.md hard rule). Returns
-    Wakeup goes through the Paperclip API (``POST /api/agents/{id}/wakeup``), never a
+    ``{"status": "ok", "issue_id"(=child), "ceo_id", ...}`` or
-    direct DB insert (CLAUDE.md hard rule). Returns
+    ``{"status": "skipped", "reason": ...}``.
-    ``{"status": "ok"|"skipped", ...}``.
     """
     if not PAPERCLIP_BOARD_API_KEY:
         logger.warning("PAPERCLIP_BOARD_API_KEY not set — skipping CEO action wakeup")
@@ -1287,17 +1538,42 @@ async def wake_ceo_for_action(
         logger.warning("No Paperclip issues found for case %s — skipping CEO action", case_number)
         return {"status": "skipped", "reason": "no_issue"}
 
-    # The main case issue — prefer the in-progress one, else the canonical
-    # "[ערר {case_number}]" issue, else the oldest. (Same selection spirit as
-    # wake_curator_for_final, but we never spawn a child — it's a side-quest.)
     main_issue = (
         next((i for i in issues if i.get("status") == "in_progress"), None)
         or next((i for i in issues if f"[ערר {case_number}]" in (i.get("title") or "")), None)
         or issues[0]
     )
     main_issue_id = main_issue["id"]
-
     ceo_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)
+
+    # Child issue assigned to the CEO — so the queued run is owned by the CEO and
+    # not cancelled by the human assignee on the parent (#227).
+    title, description = _ceo_action_brief(action, case_number)
+    child_resp = await pc_request(
+        "POST",
+        f"/api/issues/{main_issue_id}/children",
+        json={
+            "title": title,
+            "description": description,
+            "status": "in_progress",
+            "priority": "medium",
+            "assigneeAgentId": ceo_id,
+        },
+        raise_on_error=True,
+    )
+    sub_issue = child_resp.json()
+    sub_issue_id = sub_issue["id"]
+
+    # Tag plugin_state so the case page surfaces this sub-issue too.
+    try:
+        conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+        try:
+            await _link_case_to_issue(conn, sub_issue_id, case_number)
+        finally:
+            await conn.close()
+    except Exception as e:
+        logger.warning("plugin_state link failed for sub_issue=%s: %s", sub_issue_id, e)
+
     wake_resp = await pc_request(
         "POST",
         f"/api/agents/{ceo_id}/wakeup",
@@ -1306,7 +1582,7 @@ async def wake_ceo_for_action(
             "triggerDetail": "manual",
             "reason": f"generate_{action}_{case_number}",
             "payload": {
-                "issueId": main_issue_id,
+                "issueId": sub_issue_id,
                 "action": action,
                 "case_number": case_number,
             },
@@ -1314,17 +1590,85 @@ async def wake_ceo_for_action(
         raise_on_error=True,
     )
     logger.info(
-        "CEO action wakeup for case %s: action=%s issue=%s ceo=%s status=%s",
+        "CEO action wakeup for case %s: action=%s child_issue=%s ceo=%s status=%s",
-        case_number, action, main_issue_id, ceo_id, wake_resp.status_code,
+        case_number, action, sub_issue_id, ceo_id, wake_resp.status_code,
     )
     return {
         "status": "ok",
         "action": action,
-        "issue_id": main_issue_id,
+        "issue_id": sub_issue_id,
+        "main_issue_id": main_issue_id,
         "ceo_id": ceo_id,
     }
 
 
+# Substring of the generation child-issue title per action (see _ceo_action_brief).
+GENERATION_TITLE_SUBSTR = {
+    "party_claims_summary": "סיכום-מנהלים",
+    "interim_draft": "טיוטת טענות-הצדדים",
+}
+
+
+async def get_generation_run_status(case_number: str, action: str) -> dict:
+    """Server-derived run status for a "הפקת מסמכים" generation action (#227 ג).
+
+    Reconstructs the state of the most recent generation child issue for
+    (case, action) from Paperclip — the child issue + its latest heartbeat_run —
+    so the UI indicator survives navigation/reload (it never depends on the
+    browser having stayed on the page). Read-only Paperclip DB access, behind the
+    platform port (G12).
+
+    Returns a serializable dict::
+
+        {"issue_id", "issue_status", "run_status", "started_at", "finished_at",
+         "error", "error_code"}
+
+    with all timestamps as ISO-8601 UTC strings (or None). ``issue_id`` is None
+    when no generation child issue exists yet (→ the endpoint reports 'idle').
+    """
+    substr = GENERATION_TITLE_SUBSTR.get(action)
+    if not substr:
+        return {"issue_id": None}
+
+    def _iso(dt) -> str | None:
+        return dt.isoformat() if dt is not None else None
+
+    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+    try:
+        issue = await conn.fetchrow(
+            """SELECT id, status
+               FROM issues
+               WHERE title LIKE ('%' || $1 || '%') AND title LIKE ('%' || $2 || '%')
+               ORDER BY created_at DESC
+               LIMIT 1""",
+            case_number, substr,
+        )
+        if issue is None:
+            return {"issue_id": None}
+
+        run = await conn.fetchrow(
+            """SELECT hr.status, hr.started_at, hr.finished_at, hr.error, hr.error_code
+               FROM heartbeat_runs hr
+               JOIN agent_wakeup_requests awr ON hr.wakeup_request_id = awr.id
+               WHERE awr.payload->>'issueId' = $1
+               ORDER BY hr.created_at DESC
+               LIMIT 1""",
+            str(issue["id"]),
+        )
+    finally:
+        await conn.close()
+
+    return {
+        "issue_id": str(issue["id"]),
+        "issue_status": issue["status"],
+        "run_status": run["status"] if run else None,
+        "started_at": _iso(run["started_at"]) if run else None,
+        "finished_at": _iso(run["finished_at"]) if run else None,
+        "error": (run["error"] if run else None) or None,
+        "error_code": (run["error_code"] if run else None) or None,
+    }
+
+
 def _curator_task_brief(task: str, case_number: str, final_filename: str) -> tuple[str, str]:
     """Build the (sub-issue title, description) for a staged final-decision task.
 
@@ -1684,3 +2028,109 @@ async def wake_analyst_for_argument_aggregation(
         "analyst_id": analyst_id,
         "main_issue_id": main_issue_id,
     }
+
+
+async def wake_analyst_for_protocol_analysis(
+    case_number: str,
+    company_id: str,
+    document_id: str = "",
+) -> dict:
+    """Wake the legal-analyst to run the hearing-protocol comparative analysis.
+
+    Triggered by the chair clicking "נתח את פרוטוקול הדיון" / "נתח מחדש" in the
+    "מה קרה בדיון" panel (#226). Same delegation shape as
+    ``wake_analyst_for_argument_aggregation``: the FastAPI container cannot run
+    ``analyze_protocol`` directly (it calls ``claude_session.query_json()``,
+    which needs the host-side ``claude`` CLI), so we create a child issue under
+    the case's main Paperclip issue, assign it to the company's analyst, and
+    trigger a wakeup. The analyst runs the MCP tool locally and reports back.
+
+    ``document_id`` optionally pins the analysis to a specific protocol document
+    (when a case holds several protocols — #223).
+
+    Returns a dict shaped for the FastAPI endpoint to serialize as-is:
+        {"status": "queued", "sub_issue_id", "analyst_id", "main_issue_id"}
+      or {"status": "skipped", "reason": "..."} for non-fatal early outs.
+    """
+    if not PAPERCLIP_BOARD_API_KEY:
+        logger.warning(
+            "PAPERCLIP_BOARD_API_KEY not set — cannot queue analyst wakeup "
+            "for protocol analysis on %s",
+            case_number,
+        )
+        return {"status": "skipped", "reason": "no_api_key"}
+
+    analyst_id = ANALYST_AGENTS.get(company_id)
+    if not analyst_id:
+        logger.info("No analyst configured for company %s — skipping", company_id)
+        return {"status": "skipped", "reason": "no_analyst", "company_id": company_id}
+
+    issues = await get_case_issues(case_number)
+    if not issues:
+        logger.warning(
+            "No Paperclip issues found for case %s — cannot queue analyst", case_number,
+        )
+        return {"status": "skipped", "reason": "no_issue"}
+
+    main_issue = next((i for i in issues if i.get("status") == "in_progress"), None) or issues[0]
+    main_issue_id = main_issue["id"]
+
+    doc_clause = f", document_id=\"{document_id}\"" if document_id.strip() else ""
+    description = (
+        f"חיים ביקש ניתוח פרוטוקול-דיון בתיק {case_number}.\n\n"
+        f"הרץ `mcp__legal-ai__analyze_protocol(case_number=\"{case_number}\"{doc_clause})` "
+        f"וכתוב comment בעברית עם תוצאת הניתוח — כמה טענות התחזקו בדיון, כמה נטענו "
+        f"לראשונה, וכמה ירדו. אם אין פרוטוקול ועדת-ערר בתיק או שאין טיעונים מאוגדים "
+        f"להשוואה, דווח ב-comment מה חסר וסגור את ה-issue כ-blocked."
+    )
+    child_resp = await pc_request(
+        "POST",
+        f"/api/issues/{main_issue_id}/children",
+        json={
+            "title": f"[ערר {case_number}] ניתוח פרוטוקול-דיון",
+            "description": description,
+            "status": "in_progress",
+            "priority": "medium",
+            "assigneeAgentId": analyst_id,
+        },
+        raise_on_error=True,
+    )
+    sub_issue = child_resp.json()
+    sub_issue_id = sub_issue["id"]
+
+    # Tag plugin_state so the case page surfaces this sub-issue too.
+    try:
+        conn = await asyncpg.connect(PAPERCLIP_DB_URL)
+        try:
+            await _link_case_to_issue(conn, sub_issue_id, case_number)
+        finally:
+            await conn.close()
+    except Exception as e:
+        logger.warning("plugin_state link failed for sub_issue=%s: %s", sub_issue_id, e)
+
+    wake_resp = await pc_request(
+        "POST",
+        f"/api/agents/{analyst_id}/wakeup",
+        json={
+            "source": "on_demand",
+            "triggerDetail": "manual",
+            "reason": f"analyze_protocol_{case_number}",
+            "payload": {
+                "issueId": sub_issue_id,
+                "mutation": "assignment",
+                "caseNumber": case_number,
+            },
+        },
+        raise_on_error=True,
+    )
+    logger.info(
+        "Analyst wakeup for protocol analysis on case %s: sub_issue=%s "
+        "analyst=%s doc=%s wake=%s",
+        case_number, sub_issue_id, analyst_id, document_id or "auto", wake_resp.status_code,
+    )
+    return {
+        "status": "queued",
+        "sub_issue_id": sub_issue_id,
+        "analyst_id": analyst_id,
+        "main_issue_id": main_issue_id,
+    }

web/tests/test_agent_health.py

@@ -0,0 +1,145 @@
+"""Tests for #222 — agent health taxonomy (pure classifier + Paperclip fetch).
+
+Pure-classifier tests need no deps. The fetch test fakes an asyncpg connection
+(routing the three queries by content) so it never touches the live DB.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # web/
+
+from agent_health import (  # noqa: E402
+    STALL_WAKEUP_THRESHOLD,
+    classify_issue_health,
+    is_recovery_reason,
+    zombie_escalation,
+)
+from agent_health import ZOMBIE_ESCALATE_MIN_RECOVERY  # noqa: E402
+
+
+# ── pure classifier ────────────────────────────────────────────────────────
+def test_live_run_is_working_regardless_of_wakeups():
+    assert classify_issue_health(has_live_run=True, recovery_wakeups=9, total_wakeups=9) == "working"
+
+
+def test_recovery_wakeup_without_live_run_is_zombie():
+    # the stranded-child signature: assigned+open, nothing running, recovery firing
+    assert classify_issue_health(has_live_run=False, recovery_wakeups=1, total_wakeups=1) == "zombie"
+
+
+def test_repeated_wakeups_no_recovery_is_stalled():
+    assert (
+        classify_issue_health(
+            has_live_run=False, recovery_wakeups=0, total_wakeups=STALL_WAKEUP_THRESHOLD
+        )
+        == "stalled"
+    )
+
+
+def test_quiet_open_issue_is_idle():
+    assert classify_issue_health(has_live_run=False, recovery_wakeups=0, total_wakeups=1) == "idle"
+
+
+# ── watchdog escalation policy (pure) ───────────────────────────────────────
+def _item(health, recovery):
+    return {"health": health, "recovery_wakeups": recovery, "identifier": "CMP-141", "agent_name": "writer"}
+
+
+def test_zombie_escalation_medium_for_fresh_loop():
+    decision = zombie_escalation(_item("zombie", ZOMBIE_ESCALATE_MIN_RECOVERY))
+    assert decision is not None
+    severity, reason = decision
+    assert severity == "medium"
+    assert "CMP-141" in reason and "recovery" in reason.lower()
+
+
+def test_zombie_escalation_high_for_long_loop():
+    severity, _ = zombie_escalation(_item("zombie", 5))
+    assert severity == "high"
+
+
+def test_zombie_below_threshold_is_not_escalated():
+    assert zombie_escalation(_item("zombie", ZOMBIE_ESCALATE_MIN_RECOVERY - 1)) is None
+
+
+def test_non_zombie_never_escalates():
+    assert zombie_escalation(_item("stalled", 9)) is None
+    assert zombie_escalation(_item("idle", 0)) is None
+    assert zombie_escalation(_item("working", 9)) is None
+
+
+def test_is_recovery_reason_markers():
+    assert is_recovery_reason("source_scoped_recovery_action")
+    assert is_recovery_reason("issue_continuation_needed")
+    assert is_recovery_reason("issue_reopened_via_comment")
+    assert not is_recovery_reason("start_workflow_8125-09-24")
+    assert not is_recovery_reason(None)
+
+
+# ── Paperclip fetch (fake DB) ───────────────────────────────────────────────
+os.environ.setdefault("PAPERCLIP_DB_URL", "postgres://x:x@127.0.0.1:54329/paperclip")
+pc = pytest.importorskip("paperclip_client", reason="web deps unavailable")
+
+
+class _FakeConn:
+    def __init__(self, issues, live, wakeups):
+        self._issues, self._live, self._wakeups = issues, live, wakeups
+
+    async def fetch(self, query, *args):
+        if "FROM issues" in query:
+            return self._issues
+        if "heartbeat_runs" in query:
+            return self._live
+        if "agent_wakeup_requests" in query:
+            return self._wakeups
+        raise AssertionError(f"unexpected query: {query[:40]}")
+
+    async def close(self):
+        pass
+
+
+def test_get_agent_health_classifies_and_sorts(monkeypatch):
+    # zombie agent A (recovery wake, no live run); working agent B (live run);
+    # idle agent C (quiet).
+    issues = [
+        {"id": "iA", "identifier": "CMP-1", "status": "blocked", "assignee_agent_id": "agA", "agent_name": "writer"},
+        {"id": "iB", "identifier": "CMP-2", "status": "in_progress", "assignee_agent_id": "agB", "agent_name": "analyst"},
+        {"id": "iC", "identifier": "CMP-3", "status": "todo", "assignee_agent_id": "agC", "agent_name": "qa"},
+    ]
+    live = [{"agent_id": "agB"}]
+    wakeups = [
+        {"agent_id": "agA", "issue_id": "iA", "reason": "source_scoped_recovery_action"},
+        {"agent_id": "agC", "issue_id": "iC", "reason": "start_workflow_x"},
+    ]
+
+    async def _connect(_url):
+        return _FakeConn(issues, live, wakeups)
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+
+    result = asyncio.run(pc.get_agent_health())
+    assert result["ok"] is True
+    by_issue = {it["issue_id"]: it["health"] for it in result["items"]}
+    assert by_issue == {"iA": "zombie", "iB": "working", "iC": "idle"}
+    assert result["counts"]["zombie"] == 1
+    assert result["counts"]["working"] == 1
+    assert result["counts"]["idle"] == 1
+    # worst-first: zombie leads
+    assert result["items"][0]["health"] == "zombie"
+    assert result["items"][0]["recovery_wakeups"] == 1
+
+
+def test_get_agent_health_empty(monkeypatch):
+    async def _connect(_url):
+        return _FakeConn([], [], [])
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+    result = asyncio.run(pc.get_agent_health())
+    assert result == {"ok": True, "items": [], "counts": {"zombie": 0, "stalled": 0, "working": 0, "idle": 0}}

web/tests/test_agent_telemetry.py

@@ -0,0 +1,104 @@
+"""Tests for #219 — web.agent_telemetry structured agent-platform events.
+
+Verifies the emitter contract the recovery-loop observability relies on:
+- ``emit`` stamps ts/v/event/outcome and drops ``None`` fields.
+- ``instrument`` lifts standard + result keys, times the call, and derives the
+  outcome (ok / noop for ``{"ok": False}`` / error on exception).
+- an exception is re-raised after the error event — the wrapped op is **never
+  silently swallowed** (constitution §6 / INV-G4).
+- the decorator is transparent: wrapped functions stay coroutine functions.
+
+Pure-stdlib module (no Paperclip/web deps), so this runs without importorskip.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # web/
+
+from agent_telemetry import SCHEMA_VERSION, emit, instrument  # noqa: E402
+import agent_telemetry  # noqa: E402
+
+
+@pytest.fixture
+def captured(monkeypatch):
+    """Intercept the sink; yield the list of emitted event dicts."""
+    events: list[dict] = []
+    monkeypatch.setattr(agent_telemetry, "_SINK", events.append)
+    return events
+
+
+def test_emit_stamps_schema_and_drops_none(captured):
+    emit("x.test", role="ceo", case_number="8125-09-24", issue_id=None)
+    (e,) = captured
+    assert e["event"] == "x.test"
+    assert e["v"] == SCHEMA_VERSION
+    assert e["outcome"] == "ok"
+    assert e["ts"].endswith("Z")
+    assert e["case_number"] == "8125-09-24"
+    assert "issue_id" not in e  # None fields dropped
+
+
+def test_instrument_success_lifts_result_keys(captured):
+    @instrument("interaction.reaped", result_keys=("cancelled",))
+    async def reap():
+        return {"ok": True, "cancelled": 3}
+
+    result = asyncio.run(reap())
+    assert result == {"ok": True, "cancelled": 3}
+    (e,) = captured
+    assert e["event"] == "interaction.reaped"
+    assert e["outcome"] == "ok"
+    assert e["cancelled"] == 3
+    assert "duration_ms" in e
+
+
+def test_instrument_ok_false_is_noop(captured):
+    @instrument("interaction.cancelled", keys=("issue_id",))
+    async def cancel(issue_id, interaction_id):
+        return {"ok": False, "error": "already resolved"}
+
+    asyncio.run(cancel("iss-1", "int-9"))
+    (e,) = captured
+    assert e["outcome"] == "noop"
+    assert e["issue_id"] == "iss-1"
+
+
+def test_instrument_reraises_after_error_event(captured):
+    @instrument("agent.wakeup", role="ceo")
+    async def wake(case_number, company_id=""):
+        raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        asyncio.run(wake("1043-04-26", company_id="cmp"))
+
+    (e,) = captured
+    assert e["outcome"] == "error"
+    assert "boom" in e["error"]
+    assert e["agent_role"] == "ceo"
+    assert e["case_number"] == "1043-04-26"
+    assert e["company_id"] == "cmp"
+
+
+def test_decorator_is_transparent():
+    @instrument("agent.wakeup", role="analyst")
+    async def wake(case_number, company_id=""):
+        return {"ok": True}
+
+    assert inspect.iscoroutinefunction(wake)
+    assert wake.__name__ == "wake"
+
+
+def test_emit_never_raises_on_sink_failure(monkeypatch):
+    def boom(_payload):
+        raise ValueError("sink down")
+
+    monkeypatch.setattr(agent_telemetry, "_SINK", boom)
+    # Must not propagate — telemetry is best-effort, never fatal to the op.
+    emit("agent.wakeup", role="ceo")

web/tests/test_escalate_issue.py

@@ -0,0 +1,118 @@
+"""Tests for #218 — paperclip_client.escalate_issue (loop-safe chair escalation).
+
+Verifies the primitive that replaces the fragile manual PATCH dance:
+- invalid severity is rejected before any DB work.
+- the happy path performs ONE atomic human-owned transition
+  (``status='in_review'``, ``assignee_user_id=CHAIM_USER_ID``) and records a
+  ``author_type='system'`` severity note carrying the reason.
+- **no wakeup / REST call is issued** — escalation hands off to the human, it
+  must never re-invoke an agent (that is what caused the recovery loops).
+- a missing issue reports ``ok:False`` and writes no comment.
+
+Uses a fake asyncpg connection — never touches the live Paperclip DB.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("PAPERCLIP_DB_URL", "postgres://x:x@127.0.0.1:54329/paperclip")
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # web/
+
+pc = pytest.importorskip("paperclip_client", reason="web deps unavailable")
+
+
+class _FakeTxn:
+    async def __aenter__(self):
+        return None
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+class _FakeConn:
+    def __init__(self, row):
+        self._row = row
+        self.updates: list[tuple] = []
+        self.inserts: list[tuple] = []
+        self.closed = False
+
+    def transaction(self):
+        return _FakeTxn()
+
+    async def fetchrow(self, query, *args):
+        self.updates.append((query, args))
+        return self._row
+
+    async def execute(self, query, *args):
+        self.inserts.append((query, args))
+
+    async def close(self):
+        self.closed = True
+
+
+@pytest.fixture
+def fake_db(monkeypatch):
+    """Patch asyncpg.connect + guard that no wakeup/REST is issued."""
+    conns: list[_FakeConn] = []
+    row = {"id": "iss-uuid", "identifier": "CMP-141", "company_id": "cmp-uuid"}
+    holder = {"row": row}
+
+    async def _connect(_url):
+        conn = _FakeConn(holder["row"])
+        conns.append(conn)
+        return conn
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+
+    async def _forbidden(*a, **k):  # any REST/wakeup would be a loop-risk
+        raise AssertionError("escalate_issue must not issue a wakeup/REST call")
+
+    monkeypatch.setattr(pc, "pc_request", _forbidden, raising=False)
+    return {"conns": conns, "holder": holder}
+
+
+def test_invalid_severity_rejected_before_db(fake_db):
+    result = asyncio.run(pc.escalate_issue("iss-1", "urgent", "boom"))
+    assert result["ok"] is False
+    assert "invalid severity" in result["error"]
+    assert fake_db["conns"] == []  # never connected
+
+
+def test_happy_path_atomic_transition_and_system_note(fake_db):
+    result = asyncio.run(
+        pc.escalate_issue("iss-uuid", "high", "analyst wedged on protocol parse")
+    )
+    assert result == {
+        "ok": True,
+        "id": "iss-uuid",
+        "identifier": "CMP-141",
+        "severity": "high",
+        "status": "in_review",
+    }
+    conn = fake_db["conns"][0]
+    # 1) one human-owned UPDATE with the chair user + in_review
+    (update_sql, update_args) = conn.updates[0]
+    assert "status='in_review'" in update_sql
+    assert "assignee_agent_id=null" in update_sql
+    assert update_args[0] == pc.CHAIM_USER_ID
+    assert update_args[1] == "iss-uuid"
+    # 2) a system-authored severity note carrying the reason
+    (insert_sql, insert_args) = conn.inserts[0]
+    assert "issue_comments" in insert_sql
+    assert "'system'" in insert_sql
+    assert "high" in insert_args[3] and "wedged on protocol parse" in insert_args[3]
+    assert conn.closed is True
+
+
+def test_missing_issue_reports_not_found_and_writes_no_comment(fake_db):
+    fake_db["holder"]["row"] = None
+    result = asyncio.run(pc.escalate_issue("ghost", "medium", "nope"))
+    assert result["ok"] is False and "not found" in result["error"]
+    conn = fake_db["conns"][0]
+    assert conn.inserts == []  # no comment on a non-existent issue

web/tests/test_predecessor_context.py

@@ -0,0 +1,84 @@
+"""Tests for #220 — paperclip_client.get_predecessor_context ("seance").
+
+Fakes an asyncpg connection so it never touches the live DB. Verifies the
+limit clamp, row shaping (timestamps → isoformat), and the empty case.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("PAPERCLIP_DB_URL", "postgres://x:x@127.0.0.1:54329/paperclip")
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # web/
+
+pc = pytest.importorskip("paperclip_client", reason="web deps unavailable")
+
+
+class _FakeConn:
+    def __init__(self, rows):
+        self._rows = rows
+        self.calls: list[tuple] = []
+
+    async def fetch(self, query, *args):
+        self.calls.append((query, args))
+        return self._rows
+
+    async def close(self):
+        pass
+
+
+def _row():
+    return {
+        "id": "run-1",
+        "status": "succeeded",
+        "started_at": datetime(2026, 7, 5, 10, 0, 0),
+        "finished_at": datetime(2026, 7, 5, 10, 5, 0),
+        "summary": "כתבתי בלוק ז — 3,206 מילים",
+        "error_code": None,
+        "session_id_after": "sess-abc",
+        "agent_name": "writer",
+    }
+
+
+@pytest.fixture
+def fake(monkeypatch):
+    holder = {"rows": [_row()], "conn": None}
+
+    async def _connect(_url):
+        holder["conn"] = _FakeConn(holder["rows"])
+        return holder["conn"]
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+    return holder
+
+
+def test_shapes_runs_newest_first(fake):
+    result = asyncio.run(pc.get_predecessor_context("iss-1"))
+    assert result["ok"] is True and result["issue_id"] == "iss-1"
+    (run,) = result["runs"]
+    assert run["run_id"] == "run-1"
+    assert run["status"] == "succeeded"
+    assert run["started_at"] == "2026-07-05T10:00:00"
+    assert run["finished_at"] == "2026-07-05T10:05:00"
+    assert "בלוק ז" in run["summary"]
+    assert run["session_id"] == "sess-abc"
+    assert run["agent_name"] == "writer"
+
+
+def test_limit_is_clamped(fake):
+    asyncio.run(pc.get_predecessor_context("iss-1", limit=99))
+    assert fake["conn"].calls[0][1] == ("iss-1", 10)  # clamped high
+    asyncio.run(pc.get_predecessor_context("iss-1", limit=0))
+    assert fake["conn"].calls[0][1] == ("iss-1", 1)  # clamped low
+
+
+def test_empty(fake):
+    fake["rows"] = []
+    result = asyncio.run(pc.get_predecessor_context("ghost"))
+    assert result == {"ok": True, "issue_id": "ghost", "runs": []}