legal-ai/web/paperclip_client.py

"""Paperclip integration via direct DB access (embedded PostgreSQL).

Creates projects, issues, and plugin state entries to fully link
a legal-ai case into Paperclip's workflow.
"""

from __future__ import annotations

import json
import logging
import os
import uuid

import asyncpg

from web.paperclip_api import pc_request, require_paperclip_db_url

logger = logging.getLogger(__name__)

# INV-ENV4 / GAP-57: no hard-coded credential default — fail loud if unset.
PAPERCLIP_DB_URL = require_paperclip_db_url()

PLUGIN_ID = "53461b5a-7f58-411a-9952-72f9c8d4a328"  # marcusgroup.legal-ai
# PAPERCLIP_API_URL — moved to web.paperclip_api (used only by pc_request now).
# Direct DB calls below use PAPERCLIP_DB_URL instead.
PAPERCLIP_BOARD_API_KEY = os.environ.get("PAPERCLIP_BOARD_API_KEY", "")

# Default workspace attached to every new Paperclip project — points agents at
# the legal-ai source tree on the host. Override via env if the path differs.
LEGAL_AI_WORKSPACE_CWD = os.environ.get("LEGAL_AI_WORKSPACE_CWD", "/home/chaim/legal-ai")
LEGAL_AI_WORKSPACE_NAME = os.environ.get("LEGAL_AI_WORKSPACE_NAME", "legal-ai")

# Company IDs from Paperclip DB
COMPANIES = {
    "licensing": "42a7acd0-30c5-4cbd-ac97-7424f65df294",   # CMP — רישוי ובניה
    "betterment": "8639e837-4c9d-47fa-a76b-95788d651896",  # CMPA — היטלי השבחה
}

# CEO agent per company — used for wakeup routing
CEO_AGENTS = {
    COMPANIES["licensing"]: "752cebdd-6748-4a04-aacd-c7ab0294ef33",   # CMP CEO
    COMPANIES["betterment"]: "cdbfa8bc-3d61-41a4-a2e7-677ec7d34562",  # CMPA CEO
}
# Default for backwards compat
CEO_AGENT_ID = CEO_AGENTS[COMPANIES["licensing"]]

# Knowledge Curator (Hermes) agent per company — woken after a case is
# marked final, to analyze the signed decision and propose updates to
# the style guide / lessons. POC stage 1.
CURATOR_AGENTS = {
    COMPANIES["licensing"]: "60dce831-5c5b-4bae-bda9-5282d506f0dc",   # CMP curator
    COMPANIES["betterment"]: "d6f7c55d-570a-46b8-8d72-1286d07da0d8",  # CMPA curator
}

# Legal Analyst (מנתח משפטי) agent per company — woken from the chair UI
# when the chair finishes tagging appraisals and asks for fact extraction.
# The analyst runs `mcp__legal-ai__extract_appraiser_facts` locally (where
# the Claude CLI is present), since the FastAPI container cannot.
ANALYST_AGENTS = {
    COMPANIES["licensing"]: "c26e9439-a88a-49dc-9e67-2262c95db65c",   # CMP analyst
    COMPANIES["betterment"]: "f70fd353-6cde-46b3-8d6c-cfad12100b1b",  # CMPA analyst
}

# Fallback mapping — used only when DB lookup returns no results.
# בל"מ (extension_request_*) variants route to the same company as their
# parent domain — בל"מ ברישוי → CMP, בל"מ בהיטל השבחה → CMPA, וכו'.
_FALLBACK_APPEAL_TYPE_TO_COMPANY = {
    "רישוי": COMPANIES["licensing"],
    "היטל השבחה": COMPANIES["betterment"],
    "פיצויים": COMPANIES["betterment"],
    "building_permit": COMPANIES["licensing"],
    "betterment_levy": COMPANIES["betterment"],
    "compensation_197": COMPANIES["betterment"],
    "compensation": COMPANIES["betterment"],
    "licensing": COMPANIES["licensing"],
    # בל"מ subtypes — route per domain
    "extension_request_building_permit": COMPANIES["licensing"],
    "extension_request_betterment_levy": COMPANIES["betterment"],
    "extension_request_compensation": COMPANIES["betterment"],
}

# Legal-AI DB URL for reading tag_company_mappings
_LEGAL_DB_URL = os.environ.get("POSTGRES_URL") or os.environ.get(
    "DATABASE_URL", "postgresql://legal:legal@127.0.0.1:5432/legal_ai"
)


async def _get_company_id(appeal_type: str) -> str:
    """Resolve appeal_type tag to a Paperclip company ID via DB mappings, with fallback."""
    try:
        conn = await asyncpg.connect(_LEGAL_DB_URL)
        try:
            row = await conn.fetchrow(
                "SELECT company_id FROM tag_company_mappings WHERE tag = $1 LIMIT 1",
                appeal_type,
            )
            if row:
                return row["company_id"]
        finally:
            await conn.close()
    except Exception:
        logger.debug("DB lookup for tag mapping failed, using fallback for '%s'", appeal_type)

    return _FALLBACK_APPEAL_TYPE_TO_COMPANY.get(appeal_type, COMPANIES["licensing"])


async def create_project(
    case_number: str,
    title: str,
    description: str = "",
    appeal_type: str = "רישוי",
    color: str = "#6366f1",
) -> dict:
    """Create a project in the Paperclip embedded DB, or return existing one."""
    company_id = await _get_company_id(appeal_type)

    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        # Resolve prefix from company issue_prefix in Paperclip DB
        comp_row = await conn.fetchrow(
            "SELECT issue_prefix FROM companies WHERE id = $1::uuid", company_id,
        )
        prefix = comp_row["issue_prefix"] if comp_row and comp_row["issue_prefix"] else "CMP"

        # Check for existing project with this case number
        existing = await conn.fetchrow(
            "SELECT id, name FROM projects WHERE name LIKE $1 AND company_id = $2::uuid",
            f"%{case_number}%", company_id,
        )
        if existing:
            # Backfill: ensure legacy projects also have a default workspace.
            await _ensure_default_workspace(conn, str(existing["id"]), company_id)
            return {
                "id": str(existing["id"]),
                "company_id": company_id,
                "name": existing["name"],
                "url": f"https://pc.nautilus.marcusgroup.org/{prefix}/projects/{existing['id']}/issues",
                "existing": True,
            }

        project_id = str(uuid.uuid4())
        project_name = f"ערר {case_number} — {title}"[:200]
        await conn.execute(
            """INSERT INTO projects (id, company_id, name, description, status, color)
               VALUES ($1, $2::uuid, $3, $4, 'backlog', $5)""",
            project_id, company_id, project_name, description[:500] if description else "", color,
        )
        # Default primary workspace — points agents at the legal-ai source tree.
        await _ensure_default_workspace(conn, project_id, company_id)
        # Create initial issue linked to the project
        issue_id, identifier = await _create_issue(
            conn, company_id, project_id, case_number, title, prefix,
        )

        # Link issue to legal-ai case via plugin state
        await _link_case_to_issue(conn, issue_id, case_number)

        # Verify project creation and close the setup issue
        await _verify_and_close_setup_issue(conn, project_id, issue_id, identifier, case_number)

        return {
            "id": project_id,
            "company_id": company_id,
            "name": project_name,
            "issue_id": issue_id,
            "issue_identifier": identifier,
            "url": f"https://pc.nautilus.marcusgroup.org/{prefix}/projects/{project_id}/issues",
            "existing": False,
        }
    finally:
        await conn.close()


async def archive_project(case_number: str) -> dict:
    """Set archived_at on the Paperclip project matching this case number,
    and cancel any open issues so the legal-ai UI's agent widget stops
    reporting "agents are working" on a closed case.

    The project is identified by `name LIKE '%{case_number}%'` (consistent with
    `create_project`'s lookup). Idempotent — re-archiving a project that's
    already archived returns the existing timestamp without re-cancelling
    issues that have already been completed.
    """
    # Issue statuses considered "open" — anything not done/cancelled.
    OPEN_STATUSES = ("backlog", "todo", "in_progress", "blocked", "in_review")

    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        row = await conn.fetchrow(
            """UPDATE projects
               SET archived_at = COALESCE(archived_at, now()),
                   updated_at = now()
               WHERE name LIKE $1 RETURNING id, name, archived_at""",
            f"%{case_number}%",
        )
        if not row:
            return {"status": "not_found", "case_number": case_number}

        cancelled = await conn.fetch(
            """UPDATE issues
               SET status = 'cancelled',
                   cancelled_at = now(),
                   updated_at = now()
               WHERE project_id = $1 AND status = ANY($2::text[])
               RETURNING identifier, title""",
            row["id"], list(OPEN_STATUSES),
        )

        return {
            "status": "archived",
            "project_id": str(row["id"]),
            "name": row["name"],
            "archived_at": row["archived_at"].isoformat() if row["archived_at"] else None,
            "issues_cancelled": [
                {"identifier": r["identifier"], "title": r["title"]}
                for r in cancelled
            ],
        }
    finally:
        await conn.close()


async def restore_project(case_number: str) -> dict:
    """Clear archived_at on the Paperclip project matching this case number.

    Idempotent — if already active, returns success without changes.
    """
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        row = await conn.fetchrow(
            """UPDATE projects
               SET archived_at = NULL,
                   updated_at = now()
               WHERE name LIKE $1 RETURNING id, name""",
            f"%{case_number}%",
        )
        if not row:
            return {"status": "not_found", "case_number": case_number}
        return {
            "status": "restored",
            "project_id": str(row["id"]),
            "name": row["name"],
        }
    finally:
        await conn.close()


async def update_project_name(case_number: str, new_title: str) -> None:
    """Update the Paperclip project name when a case title changes."""
    project_name = f"ערר {case_number} — {new_title}"[:200]
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        await conn.execute(
            "UPDATE projects SET name = $1, updated_at = now() WHERE name LIKE $2",
            project_name, f"%{case_number}%",
        )
    except Exception:
        logger.warning("Failed to update Paperclip project name for case %s", case_number)
    finally:
        await conn.close()


async def _ensure_default_workspace(
    conn: asyncpg.Connection,
    project_id: str,
    company_id: str,
) -> None:
    """Idempotently attach a primary workspace to the project so the
    "סביבות עבודה" tab appears in the Paperclip UI and agents wake up with
    cwd=`/home/chaim/legal-ai`. No-op if any workspace already exists.
    """
    existing = await conn.fetchval(
        "SELECT id FROM project_workspaces WHERE project_id = $1::uuid LIMIT 1",
        project_id,
    )
    if existing:
        return

    await conn.execute(
        """INSERT INTO project_workspaces
               (company_id, project_id, name, cwd, is_primary, source_type, visibility)
           VALUES ($1::uuid, $2::uuid, $3, $4, TRUE, 'local_path', 'default')""",
        company_id, project_id, LEGAL_AI_WORKSPACE_NAME, LEGAL_AI_WORKSPACE_CWD,
    )
    logger.info(
        "Attached default workspace (cwd=%s) to project %s",
        LEGAL_AI_WORKSPACE_CWD, project_id,
    )


async def _create_issue(
    conn: asyncpg.Connection,
    company_id: str,
    project_id: str,
    case_number: str,
    title: str,
    prefix: str,
) -> tuple[str, str]:
    """Create an issue in the project and return (issue_id, identifier)."""
    issue_id = str(uuid.uuid4())

    # Get next issue number for this company
    row = await conn.fetchrow(
        "UPDATE companies SET issue_counter = issue_counter + 1 WHERE id = $1::uuid RETURNING issue_counter",
        company_id,
    )
    issue_number = row["issue_counter"]
    identifier = f"{prefix}-{issue_number}"

    # Assign to the company's CEO so Paperclip's wakeup gate
    # (heartbeat staleness check) accepts the wakeup. Without this,
    # the run is cancelled with "issue assignee changed before the
    # queued run could start" and the agent never starts.
    ceo_agent_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)

    await conn.execute(
        """INSERT INTO issues (id, company_id, project_id, title, description, status, priority, issue_number, identifier, assignee_agent_id)
           VALUES ($1, $2::uuid, $3::uuid, $4, $5, 'todo', 'medium', $6, $7, $8::uuid)""",
        issue_id, company_id, project_id,
        f"[ערר {case_number}] {title}"[:200],
        f"תיק ערר {case_number}\nנוצר אוטומטית מממשק העלאת מסמכים",
        issue_number, identifier, ceo_agent_id,
    )

    logger.info("Created Paperclip issue %s: [ערר %s] %s", identifier, case_number, title)
    return issue_id, identifier


async def _link_case_to_issue(conn: asyncpg.Connection, issue_id: str, case_number: str) -> None:
    """Store the legal-ai case number in plugin state, linked to the issue."""
    await conn.execute(
        """INSERT INTO plugin_state (plugin_id, scope_kind, scope_id, namespace, state_key, value_json)
           VALUES ($1::uuid, 'issue', $2, 'default', 'legal-case-number', $3::jsonb)
           ON CONFLICT (plugin_id, scope_kind, scope_id, namespace, state_key) DO UPDATE SET value_json = $3::jsonb""",
        PLUGIN_ID, issue_id, json.dumps(case_number),
    )
    logger.info("Linked issue %s to case %s via plugin state", issue_id, case_number)


async def _verify_and_close_setup_issue(
    conn: asyncpg.Connection,
    project_id: str,
    issue_id: str,
    identifier: str,
    case_number: str,
) -> None:
    """Verify the project was created correctly, then transition the setup issue to done."""
    # Move to in_progress while verifying
    await conn.execute(
        "UPDATE issues SET status = 'in_progress', started_at = now() WHERE id = $1",
        issue_id,
    )
    logger.info("%s: בביצוע — מאמת יצירת פרויקט", identifier)

    # Verify: project exists, issue is linked, plugin state exists
    checks = []

    project = await conn.fetchrow("SELECT id, name FROM projects WHERE id = $1::uuid", project_id)
    checks.append(("פרויקט נוצר", project is not None))

    issue = await conn.fetchrow(
        "SELECT id, project_id FROM issues WHERE id = $1 AND project_id = $2::uuid",
        issue_id, project_id,
    )
    checks.append(("משימה משויכת לפרויקט", issue is not None))

    plugin_link = await conn.fetchrow(
        "SELECT value_json FROM plugin_state WHERE scope_id = $1 AND state_key = 'legal-case-number'",
        issue_id,
    )
    checks.append(("קישור למערכת המשפטית", plugin_link is not None))

    all_ok = all(ok for _, ok in checks)
    report_lines = [f"{'✓' if ok else '✕'} {name}" for name, ok in checks]
    report = "\n".join(report_lines)

    if all_ok:
        await conn.execute(
            "UPDATE issues SET status = 'done', completed_at = now() WHERE id = $1",
            issue_id,
        )
        # Document the verification in a comment
        await conn.execute(
            """INSERT INTO issue_comments (id, company_id, issue_id, body)
               VALUES ($1, (SELECT company_id FROM issues WHERE id = $2), $2,
               $3)""",
            str(uuid.uuid4()), issue_id,
            f"## אימות יצירת פרויקט — ערר {case_number}\n\n{report}\n\nהפרויקט נוצר בהצלחה. משימה נסגרה אוטומטית.",
        )
        logger.info("%s: הושלם — פרויקט אומת ונסגר", identifier)
    else:
        # Leave in_progress with a warning comment
        failed = [name for name, ok in checks if not ok]
        await conn.execute(
            """INSERT INTO issue_comments (id, company_id, issue_id, body)
               VALUES ($1, (SELECT company_id FROM issues WHERE id = $2), $2,
               $3)""",
            str(uuid.uuid4()), issue_id,
            f"## אימות יצירת פרויקט — ערר {case_number}\n\n{report}\n\n⚠️ בדיקות שנכשלו: {', '.join(failed)}",
        )
        logger.warning("%s: אימות נכשל — %s", identifier, ", ".join(failed))


async def get_project_url(case_number: str) -> str | None:
    """Find existing Paperclip project for a case number."""
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        row = await conn.fetchrow(
            "SELECT id, company_id FROM projects WHERE name LIKE $1",
            f"%{case_number}%",
        )
        if row:
            comp_row = await conn.fetchrow(
                "SELECT issue_prefix FROM companies WHERE id = $1::uuid", str(row["company_id"]),
            )
            prefix = comp_row["issue_prefix"] if comp_row and comp_row["issue_prefix"] else "CMP"
            return f"https://pc.nautilus.marcusgroup.org/{prefix}/projects/{row['id']}/issues"
        return None
    finally:
        await conn.close()


async def create_workflow_issue(case_number: str, title: str) -> dict:
    """Create a workflow issue in the existing Paperclip project for a case.

    Returns dict with issue_id, identifier, project_url.
    Raises ValueError if no project found.
    """
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        # Find existing project
        row = await conn.fetchrow(
            "SELECT id, company_id FROM projects WHERE name LIKE $1",
            f"%{case_number}%",
        )
        if not row:
            raise ValueError(f"No Paperclip project found for case {case_number}")

        project_id = str(row["id"])
        company_id = str(row["company_id"])

        # Get company prefix
        comp_row = await conn.fetchrow(
            "SELECT issue_prefix FROM companies WHERE id = $1::uuid", company_id,
        )
        prefix = comp_row["issue_prefix"] if comp_row and comp_row["issue_prefix"] else "CMP"

        # Create the workflow issue
        issue_id, identifier = await _create_issue(
            conn, company_id, project_id, case_number,
            f"התחל תהליך ניסוח — {title}"[:200], prefix,
        )

        # Link to legal-ai case via plugin state
        await _link_case_to_issue(conn, issue_id, case_number)

        project_url = f"https://pc.nautilus.marcusgroup.org/{prefix}/projects/{project_id}/issues"
        logger.info("Created workflow issue %s for case %s", identifier, case_number)

        return {
            "issue_id": issue_id,
            "identifier": identifier,
            "company_id": company_id,
            "project_url": project_url,
        }
    finally:
        await conn.close()


async def get_case_issues(case_number: str) -> list[dict]:
    """Get all Paperclip issues linked to a legal-ai case number.

    Matches via two paths to avoid missing historical issues:
      (a) the original setup linkage in plugin_state (state_key = legal-case-number)
      (b) issues whose title contains "[ערר {case_number}]" or "ערר {case_number}"
          — that's how sub-agents conventionally tag follow-up issues.
    Returns the union of both, deduplicated by issue id, ordered by creation time.
    """
    title_patterns = [f"%[ערר {case_number}]%", f"%ערר {case_number}%"]
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT DISTINCT ON (i.id)
                      i.id, i.title, i.status, i.identifier, i.priority,
                      i.assignee_agent_id, a.name AS assignee_name,
                      i.started_at, i.completed_at, i.created_at, i.company_id
               FROM issues i
               LEFT JOIN agents a ON i.assignee_agent_id = a.id
               LEFT JOIN plugin_state ps ON ps.scope_id = i.id::text
                                        AND ps.plugin_id = $1::uuid
                                        AND ps.state_key = 'legal-case-number'
                                        AND ps.value_json = $2::jsonb
               WHERE ps.scope_id IS NOT NULL
                  OR i.title LIKE ANY($3::text[])
               ORDER BY i.id, i.created_at""",
            PLUGIN_ID, json.dumps(case_number), title_patterns,
        )
        # Sort by created_at after dedup
        sorted_rows = sorted(rows, key=lambda r: r["created_at"])
        return [
            {
                "id": str(r["id"]),
                "title": r["title"],
                "status": r["status"],
                "identifier": r["identifier"],
                "priority": r["priority"],
                "assignee_name": r["assignee_name"],
                "started_at": r["started_at"].isoformat() if r["started_at"] else None,
                "completed_at": r["completed_at"].isoformat() if r["completed_at"] else None,
                "created_at": r["created_at"].isoformat() if r["created_at"] else None,
                "company_id": str(r["company_id"]),
            }
            for r in sorted_rows
        ]
    finally:
        await conn.close()


async def get_issue_comments(issue_ids: list[str]) -> list[dict]:
    """Get all comments on a list of Paperclip issues, with agent metadata."""
    if not issue_ids:
        return []
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT ic.id, ic.issue_id, ic.body, ic.created_at,
                      ic.author_agent_id, ic.author_user_id,
                      a.name AS agent_name, a.role AS agent_role, a.icon AS agent_icon
               FROM issue_comments ic
               LEFT JOIN agents a ON ic.author_agent_id = a.id
               WHERE ic.issue_id = ANY($1::uuid[])
               ORDER BY ic.created_at""",
            issue_ids,
        )
        return [
            {
                "id": str(r["id"]),
                "issue_id": str(r["issue_id"]),
                "body": r["body"],
                "created_at": r["created_at"].isoformat() if r["created_at"] else None,
                "author_agent_id": str(r["author_agent_id"]) if r["author_agent_id"] else None,
                "author_user_id": r["author_user_id"],
                "agent_name": r["agent_name"],
                "agent_role": r["agent_role"],
                "agent_icon": r["agent_icon"],
            }
            for r in rows
        ]
    finally:
        await conn.close()


async def get_agents_for_company(company_id: str) -> list[dict]:
    """Get all agents belonging to a Paperclip company."""
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT id, name, role, title, status, icon, last_heartbeat_at
               FROM agents
               WHERE company_id = $1::uuid
               ORDER BY role, name""",
            company_id,
        )
        return [
            {
                "id": str(r["id"]),
                "name": r["name"],
                "role": r["role"],
                "title": r["title"],
                "status": r["status"],
                "icon": r["icon"],
                "last_heartbeat_at": r["last_heartbeat_at"].isoformat() if r["last_heartbeat_at"] else None,
            }
            for r in rows
        ]
    finally:
        await conn.close()


async def get_agents_for_case(company_id: str, issue_ids: list[str]) -> list[dict]:
    """Get agents with per-case status (running on *this* case vs globally)."""
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT a.id, a.name, a.role, a.title, a.icon,
                      a.status AS global_status, a.last_heartbeat_at,
                      EXISTS(
                        SELECT 1 FROM heartbeat_runs hr
                        JOIN agent_wakeup_requests wr ON hr.wakeup_request_id = wr.id
                        WHERE hr.agent_id = a.id
                          AND hr.status = 'running'
                          AND wr.payload->>'issueId' = ANY($2::text[])
                      ) AS active_on_case
               FROM agents a
               WHERE a.company_id = $1::uuid
               ORDER BY a.role, a.name""",
            company_id, issue_ids,
        )
        return [
            {
                "id": str(r["id"]),
                "name": r["name"],
                "role": r["role"],
                "title": r["title"],
                "status": "running" if r["active_on_case"] else (
                    "idle" if r["global_status"] == "running" else r["global_status"]
                ),
                "icon": r["icon"],
                "last_heartbeat_at": r["last_heartbeat_at"].isoformat() if r["last_heartbeat_at"] else None,
            }
            for r in rows
        ]
    finally:
        await conn.close()


async def post_comment(issue_id: str, company_id: str, body: str) -> dict:
    """Post a comment on a Paperclip issue.

    Tries the Board API first (triggers plugin events for CEO routing).
    Falls back to direct DB insert + CEO wakeup if API fails.
    """
    # Try Board API first — this triggers the event bus
    if PAPERCLIP_BOARD_API_KEY:
        try:
            resp = await pc_request(
                "POST",
                f"/api/board/issues/{issue_id}/comments",
                json={"body": body},
            )
            if resp.status_code < 400:
                result = resp.json()
                logger.info("Posted comment via Board API on issue %s", issue_id)
                return {"comment_id": result.get("id", ""), "issue_id": issue_id, "method": "api"}
        except Exception:
            logger.debug("Board API comment failed for issue %s, falling back to DB", issue_id)

    # Fallback: direct DB insert + explicit CEO wakeup
    comment_id = str(uuid.uuid4())
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        await conn.execute(
            """INSERT INTO issue_comments (id, company_id, issue_id, author_user_id, body)
               VALUES ($1, $2::uuid, $3::uuid, 'chaim', $4)""",
            comment_id, company_id, issue_id, body,
        )
        logger.info("Posted comment via DB fallback on issue %s", issue_id)
    finally:
        await conn.close()

    # Wake the correct CEO for this company
    ceo_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)
    try:
        await pc_request(
            "POST",
            f"/api/agents/{ceo_id}/wakeup",
            json={
                "source": "on_demand",
                "triggerDetail": "manual",
                "reason": f"user_comment_{issue_id}",
                "payload": {"issueId": issue_id, "mutation": "comment"},
            },
            raise_on_error=True,
        )
    except Exception:
        logger.warning("Failed to wake CEO after DB comment on issue %s", issue_id)

    return {"comment_id": comment_id, "issue_id": issue_id, "method": "db_fallback"}


async def get_issue_interactions(issue_ids: list[str]) -> list[dict]:
    """Fetch issue-thread interactions (agent → user button prompts).

    Returns all `pending` interactions plus any resolved within the last 24h
    so the user sees a brief tail of recent answers without flooding the feed.
    Ordered by ``created_at`` so callers can interleave with comments.
    """
    if not issue_ids:
        return []
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """SELECT id, issue_id, kind, status, title, summary,
                      payload, result, created_at, resolved_at
               FROM issue_thread_interactions
               WHERE issue_id = ANY($1::uuid[])
                 AND (status = 'pending'
                      OR resolved_at > now() - interval '24 hours')
               ORDER BY created_at""",
            issue_ids,
        )
        out: list[dict] = []
        for r in rows:
            payload = r["payload"]
            result = r["result"]
            if isinstance(payload, str):
                try:
                    payload = json.loads(payload)
                except Exception:
                    payload = {}
            if isinstance(result, str):
                try:
                    result = json.loads(result)
                except Exception:
                    result = None
            out.append({
                "id": str(r["id"]),
                "issue_id": str(r["issue_id"]),
                "kind": r["kind"],
                "status": r["status"],
                "title": r["title"],
                "summary": r["summary"],
                "payload": payload or {},
                "result": result,
                "created_at": r["created_at"].isoformat() if r["created_at"] else None,
                "resolved_at": r["resolved_at"].isoformat() if r["resolved_at"] else None,
            })
        return out
    finally:
        await conn.close()


# ── Agent-run observability + control ───────────────────────────────────────
# Live view of which agents are actually working right now + their output, and
# the controls to manage a stuck/runaway run. These wrap Paperclip's own
# heartbeat-run API (verified live): we use the *graceful* platform endpoints
# (cancel / reset-session) — never a raw kill on the run's process_pid, which
# would bypass the platform's watchdog and our extractors' per-chunk
# checkpointing. The only seam allowed to call these is web/agent_platform_port.

async def list_live_runs(company_id: str) -> list[dict]:
    """Queued + running heartbeat runs for a company (GET .../live-runs).

    Each row carries ``agentName``/``status``/``issueId``/``startedAt`` and an
    ``outputSilence`` block (``level`` ok|suspicion|critical) — the platform's
    own liveness signal, surfaced so the UI can flag a stalled run.
    """
    resp = await pc_request(
        "GET", f"/api/companies/{company_id}/live-runs", raise_on_error=True,
    )
    data = resp.json()
    return data if isinstance(data, list) else []


async def get_run_log(run_id: str) -> dict:
    """Full output log of a heartbeat run (GET /api/heartbeat-runs/{id}/log).

    Returns the platform payload as-is: ``{runId, store, logRef, content}``
    where ``content`` is the NDJSON stream the adapter captured.
    """
    resp = await pc_request(
        "GET", f"/api/heartbeat-runs/{run_id}/log", raise_on_error=True,
    )
    return resp.json()


async def get_run_events(run_id: str) -> list[dict]:
    """Lifecycle/event timeline of a heartbeat run (.../events)."""
    resp = await pc_request(
        "GET", f"/api/heartbeat-runs/{run_id}/events", raise_on_error=True,
    )
    data = resp.json()
    return data if isinstance(data, list) else []


async def cancel_run(run_id: str) -> dict:
    """Gracefully cancel a queued/running heartbeat run (POST .../cancel).

    The platform stops the run cleanly (process-group teardown + status flip),
    respecting the watchdog. Safe for the halacha drain: its extractor is
    checkpointed per-chunk and resumes on the next drain — a cancel loses at
    most the in-flight chunk.
    """
    resp = await pc_request(
        "POST", f"/api/heartbeat-runs/{run_id}/cancel", json={},
        raise_on_error=True,
    )
    return resp.json()


async def reset_agent_session(agent_id: str) -> dict:
    """Reset an agent's runtime session (.../runtime-state/reset-session).

    Clears a wedged session so the next wakeup starts clean — the smart
    alternative to cancelling individual runs when an agent loops.
    """
    resp = await pc_request(
        "POST", f"/api/agents/{agent_id}/runtime-state/reset-session", json={},
        raise_on_error=True,
    )
    return resp.json()


CHAIM_USER_ID = "ZpDWXxFweC3MftuF1Ttyu2VUbSPHbKZd"


async def reset_case_agents(case_number: str) -> dict:
    """Reset agent state for a case: clear error status + reassign stuck issues to user.

    Two actions:
    1. Any non-completed issue still assigned to an agent is reassigned to the chair user,
       stopping Paperclip's source_scoped_recovery_action loop.
    2. Every agent in the case's company whose global status is 'error' gets a
       reset_agent_session call (clears wedged runtime) and its DB status set to 'idle'.
    """
    first_digit = case_number.split("-")[0][0] if case_number else "1"
    company_id = COMPANIES["betterment"] if first_digit in ("8", "9") else COMPANIES["licensing"]

    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        project = await conn.fetchrow(
            "SELECT id FROM projects WHERE name LIKE $1 LIMIT 1",
            f"%{case_number}%",
        )
        if not project:
            return {"ok": False, "error": f"No Paperclip project found for {case_number}"}

        project_id = project["id"]

        reassigned = await conn.fetch(
            """UPDATE issues
               SET assignee_agent_id = null, assignee_user_id = $1, updated_at = now()
               WHERE project_id = $2
                 AND assignee_agent_id IS NOT NULL
                 AND status NOT IN ('done', 'cancelled')
               RETURNING id, identifier""",
            CHAIM_USER_ID, project_id,
        )

        error_agents = await conn.fetch(
            "SELECT id, name FROM agents WHERE company_id = $1::uuid AND status = 'error'",
            company_id,
        )

        if error_agents:
            await conn.execute(
                "UPDATE agents SET status = 'idle' WHERE id = ANY($1::uuid[])",
                [r["id"] for r in error_agents],
            )
    finally:
        await conn.close()

    reset_results = []
    for agent in error_agents:
        try:
            await reset_agent_session(str(agent["id"]))
            reset_results.append({"id": str(agent["id"]), "name": agent["name"], "ok": True})
        except Exception as e:
            logger.warning("reset_agent_session failed for %s: %s", agent["id"], e)
            reset_results.append({"id": str(agent["id"]), "name": agent["name"], "ok": False, "error": str(e)})

    return {
        "ok": True,
        "reassigned_issues": [{"id": str(r["id"]), "identifier": r["identifier"]} for r in reassigned],
        "reset_agents": reset_results,
    }


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:
    """Submit a user response to an `ask_user_questions` interaction.

    Paperclip auto-wakes the issue assignee on success
    (`queueResolvedInteractionContinuationWakeup`).
    """
    resp = await pc_request(
        "POST",
        f"/api/issues/{issue_id}/interactions/{interaction_id}/respond",
        json=payload,
        raise_on_error=True,
    )
    return resp.json()


async def accept_interaction(
    issue_id: str, interaction_id: str, payload: dict,
) -> dict:
    """Accept a `request_confirmation` or `suggest_tasks` interaction."""
    resp = await pc_request(
        "POST",
        f"/api/issues/{issue_id}/interactions/{interaction_id}/accept",
        json=payload,
        raise_on_error=True,
    )
    return resp.json()


async def reject_interaction(
    issue_id: str, interaction_id: str, payload: dict,
) -> dict:
    """Reject a `request_confirmation` or `suggest_tasks` interaction."""
    resp = await pc_request(
        "POST",
        f"/api/issues/{issue_id}/interactions/{interaction_id}/reject",
        json=payload,
        raise_on_error=True,
    )
    return resp.json()


async def cancel_interaction(issue_id: str, interaction_id: str) -> dict:
    """Dismiss a pending interaction WITHOUT waking the issue assignee.

    For stale/duplicate questions (TaskMaster #215). We flip the row to
    ``cancelled`` directly rather than calling Paperclip's resolve endpoints:
    those carry the interaction's ``wake_assignee`` continuation policy and would
    re-wake the agent. A direct status flip has no side effects — the table has
    no triggers (verified) and nothing reaps ``cancelled`` rows.
    """
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        row = await conn.fetchrow(
            """UPDATE issue_thread_interactions
                  SET status='cancelled', resolved_at=now(),
                      resolved_by_user_id=$3, updated_at=now()
                WHERE id=$1::uuid AND issue_id=$2::uuid AND status='pending'
            RETURNING id, status""",
            interaction_id, issue_id, CHAIM_USER_ID,
        )
    finally:
        await conn.close()
    if not row:
        return {
            "ok": False,
            "cancelled": False,
            "error": "interaction not found or already resolved",
        }
    return {"ok": True, "cancelled": True, "id": str(row["id"]), "status": row["status"]}


async def reap_stale_interactions() -> dict:
    """Cancel pending interactions stranded on closed issues (done/cancelled).

    Safety-net reaper (TaskMaster #215): a question on a closed issue can never
    be answered and only inflates the chair's "awaiting you" count — this was the
    dominant source of the pending pile-up. Scoped to the two legal-ai companies.
    Direct status flip — no wakeups, no side effects.
    """
    company_ids = list(COMPANIES.values())
    conn = await asyncpg.connect(PAPERCLIP_DB_URL)
    try:
        rows = await conn.fetch(
            """UPDATE issue_thread_interactions it
                  SET status='cancelled', resolved_at=now(),
                      resolved_by_user_id=$2, updated_at=now()
                 FROM issues i
                WHERE i.id = it.issue_id
                  AND it.status='pending'
                  AND it.company_id = ANY($1::uuid[])
                  AND i.status IN ('done', 'cancelled')
            RETURNING it.id""",
            company_ids, CHAIM_USER_ID,
        )
    finally:
        await conn.close()
    if rows:
        logger.info(
            "reap_stale_interactions: cancelled %d stranded pending interactions",
            len(rows),
        )
    return {"ok": True, "cancelled": len(rows)}


# 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 = "ספריית פסיקה — תור חילוץ"


async def _get_or_create_library_project(
    conn: asyncpg.Connection, company_id: str,
) -> str:
    """Return the project_id for the per-company library extraction queue,
    creating it (with a workspace pointing at legal-ai) if it doesn't exist."""
    row = await conn.fetchrow(
        "SELECT id FROM projects WHERE company_id = $1::uuid AND name = $2 LIMIT 1",
        company_id, _LIBRARY_PROJECT_NAME,
    )
    if row:
        return str(row["id"])

    project_id = str(uuid.uuid4())
    await conn.execute(
        """INSERT INTO projects (id, company_id, name, description, status, color)
           VALUES ($1, $2::uuid, $3, $4, 'backlog', $5)""",
        project_id, company_id, _LIBRARY_PROJECT_NAME,
        "תור אוטומטי לחילוץ הלכות ומטא-דאטה מפסיקה שהועלתה לספריה. כל issue "
        "כאן מייצג פסק דין יחיד שצריך לעבד — להריץ "
        "precedent_extract_metadata + precedent_extract_halachot עבור ה-case_law_id "
        "של ה-issue בלבד (לא precedent_process_pending, שמרוקן את כל התור).",
        "#a17a3a",  # gold-ish
    )
    await _ensure_default_workspace(conn, project_id, company_id)
    logger.info("Created library extraction project %s for company %s", project_id, company_id)
    return project_id


async def wake_for_precedent_extraction(
    case_law_id: str,
    citation: str,
    practice_area: str = "",
) -> dict:
    """Trigger Claude/Paperclip to run halacha+metadata extraction for a
    freshly-uploaded precedent.

    Creates a Paperclip issue under the per-company "ספריית פסיקה" project,
    assigns it to the company CEO, links the case_law_id via plugin_state,
    and wakes the CEO via the Board API. The issue tells the CEO to extract
    THIS case only — ``precedent_extract_metadata`` + ``precedent_extract_halachot``
    for the linked ``case_law_id`` — NOT ``precedent_process_pending``, which
    drains the whole historical backlog (hours) and blows the heartbeat's
    time budget (→ timeout/process_lost). The backlog is drained separately by
    the nightly ``legal-halacha-drain`` service (23:00–05:00).

    Best-effort: any failure is logged and swallowed so a partial Paperclip
    outage doesn't block the upload itself. ``request_halacha_extraction`` still
    stamps ``halacha_extraction_requested_at`` as a safety net, so the nightly
    drain picks the case up even if this wakeup fails.
    """
    if not PAPERCLIP_BOARD_API_KEY:
        logger.warning(
            "PAPERCLIP_BOARD_API_KEY not set — skipping precedent-extraction wakeup"
        )
        return {"ok": False, "skipped": "no_api_key"}

    company_id = await _get_company_id(practice_area)
    ceo_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)

    try:
        conn = await asyncpg.connect(PAPERCLIP_DB_URL)
        try:
            project_id = await _get_or_create_library_project(conn, company_id)

            # Bump issue counter & build identifier (matches _create_issue style).
            row = await conn.fetchrow(
                "UPDATE companies SET issue_counter = issue_counter + 1 "
                "WHERE id = $1::uuid RETURNING issue_counter",
                company_id,
            )
            issue_number = row["issue_counter"]
            prefix = "CMP" if company_id == COMPANIES["licensing"] else "CMPA"
            identifier = f"{prefix}-{issue_number}"

            issue_id = str(uuid.uuid4())
            short_citation = (citation[:120] + "…") if len(citation) > 120 else citation
            description = (
                f"פסק דין חדש הועלה לספריית הפסיקה.\n\n"
                f"**case_law_id:** `{case_law_id}`\n"
                f"**citation:** {citation}\n\n"
                f"---\n\n"
                f"**משימה — חלץ את הפסיקה הזו בלבד** (לא את התור):\n"
                f"1. `mcp__legal-ai__precedent_extract_metadata(case_law_id=\"{case_law_id}\")`\n"
                f"2. `mcp__legal-ai__precedent_extract_halachot(case_law_id=\"{case_law_id}\")`\n\n"
                f"⚠️ **אל תריץ** `precedent_process_pending` — הוא מרוקן את **כל** התור "
                f"ההיסטורי (מאות פסיקות, שעות) וחורג מתקציב-הזמן של ה-heartbeat → "
                f"timeout/process_lost. ריקון-הבאקלוג רץ בנפרד כשירות-לילה ייעודי "
                f"(`legal-halacha-drain`, 23:00–05:00). כאן — רק התיק הזה.\n\n"
                f"לאחר ריצה: סמן את ה-issue כ-done ופתח comment קצר עם מספר ההלכות "
                f"שחולצו ושדות המטא-דאטה שהושלמו."
            )
            await conn.execute(
                """INSERT INTO issues
                       (id, company_id, project_id, title, description,
                        status, priority, issue_number, identifier, assignee_agent_id)
                   VALUES ($1, $2::uuid, $3::uuid, $4, $5, 'todo', 'medium',
                           $6, $7, $8::uuid)""",
                issue_id, company_id, project_id,
                f"[ספרייה] חלץ הלכות: {short_citation}"[:200],
                description,
                issue_number, identifier, ceo_id,
            )

            # Link case_law_id via plugin_state so the agent can find it.
            await conn.execute(
                """INSERT INTO plugin_state
                       (plugin_id, scope_kind, scope_id, namespace, state_key, value_json)
                   VALUES ($1::uuid, 'issue', $2, 'default', 'precedent-case-law-id', $3::jsonb)
                   ON CONFLICT (plugin_id, scope_kind, scope_id, namespace, state_key)
                   DO UPDATE SET value_json = $3::jsonb""",
                PLUGIN_ID, issue_id, json.dumps(case_law_id),
            )
        finally:
            await conn.close()
    except Exception as e:
        logger.exception("wake_for_precedent_extraction: DB step failed: %s", e)
        return {"ok": False, "error": f"db: {e}"}

    # Wake the CEO. Per Paperclip rules: must use API + carry issueId in payload.
    payload = {
        "source": "automation",
        "triggerDetail": "precedent_library_upload",
        "reason": f"precedent_extraction_{identifier}",
        "payload": {
            "issueId": issue_id,
            "mutation": "precedent_extraction",
            "caseLawId": case_law_id,
        },
    }
    try:
        resp = await pc_request(
            "POST",
            f"/api/agents/{ceo_id}/wakeup",
            json=payload,
            raise_on_error=True,
        )
        result = resp.json()
        logger.info(
            "Precedent-extraction wakeup queued: issue=%s case_law_id=%s",
            identifier, case_law_id,
        )
        return {"ok": True, "issue_id": issue_id, "identifier": identifier, "wakeup": result}
    except Exception as e:
        logger.exception("wake_for_precedent_extraction: wakeup API failed: %s", e)
        return {"ok": False, "error": f"wakeup: {e}", "issue_id": issue_id, "identifier": identifier}


async def wake_ceo_for_feedback_fold(
    *,
    feedback_id: str,
    feedback_text: str,
    lesson_extracted: str,
    category: str,
    block_id: str = "",
    case_number: str = "",
    practice_area: str = "",
) -> dict:
    """Wake the CEO to fold one resolved chair-feedback lesson into the right
    knowledge file (the feedback→agent-knowledge loop).

    Mirrors ``wake_for_precedent_extraction``: creates a Paperclip issue under
    the library project carrying the lesson + a category→file routing rubric,
    then wakes the CEO (reason ``feedback_fold_<identifier>``). The CEO folds
    the lesson per the rubric and closes the issue.

    Best-effort: any failure is logged and returned as ``{"ok": False, ...}``
    so resolving the feedback never fails because Paperclip is down.
    """
    if not PAPERCLIP_BOARD_API_KEY:
        logger.warning("PAPERCLIP_BOARD_API_KEY not set — skipping feedback-fold wakeup")
        return {"ok": False, "skipped": "no_api_key"}

    # decision-lessons.md is shared across companies → either CEO works; route
    # by the case's practice area when known, else default to licensing.
    company_id = await _get_company_id(practice_area)
    ceo_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)

    # category → target file routing rubric (embedded in the issue so the CEO
    # has it even if its prompt is trimmed). Keep in sync with legal-ceo.md.
    routing = {
        "style": "skills/decision/SKILL.md",
        "wrong_structure": "docs/block-schema.md + docs/legal-decision-lessons.md",
        "missing_content": "docs/legal-decision-lessons.md",
        "factual_error": "docs/legal-decision-lessons.md",
        "wrong_tone": "docs/legal-decision-lessons.md",
        "other": "שיקול דעת — לרוב באג מערכת → משימת TaskMaster, לא לקח",
    }
    target = routing.get(category, "docs/legal-decision-lessons.md")

    try:
        conn = await asyncpg.connect(PAPERCLIP_DB_URL)
        try:
            project_id = await _get_or_create_library_project(conn, company_id)
            row = await conn.fetchrow(
                "UPDATE companies SET issue_counter = issue_counter + 1 "
                "WHERE id = $1::uuid RETURNING issue_counter",
                company_id,
            )
            issue_number = row["issue_counter"]
            prefix = "CMP" if company_id == COMPANIES["licensing"] else "CMPA"
            identifier = f"{prefix}-{issue_number}"

            issue_id = str(uuid.uuid4())
            short = (feedback_text[:90] + "…") if len(feedback_text) > 90 else feedback_text
            description = (
                f"היו\"ר סימנה הערת פידבק כ\"יושמה\" — יש לקפל את הלקח לקובץ הידע הנכון.\n\n"
                f"**feedback_id:** `{feedback_id}`\n"
                f"**קטגוריה:** {category}\n"
                f"**בלוק:** {block_id or '—'}\n"
                f"**תיק:** {case_number or '—'}\n\n"
                f"**ההערה:**\n{feedback_text}\n\n"
                f"**הלקח שהופק:**\n{lesson_extracted or '(אין — הפק לקח מההערה)'}\n\n"
                f"---\n\n"
                f"**יעד קיפול לפי הקטגוריה:** `{target}`\n\n"
                f"**משימה:**\n"
                f"1. קרא את קובץ היעד והבן מה כבר מתועד שם.\n"
                f"2. הוסף את הלקח **רק אם אינו קיים** (לא כפל). משפט אחד ברור + Rule באנגלית.\n"
                f"3. אם הקטגוריה היא `other` והלקח הוא באג מערכת ולא לקח כתיבה — "
                f"אל תוסיף לקובץ; פתח/עדכן משימת TaskMaster במקום.\n"
                f"4. סמן את ה-issue כ-done ופתח comment קצר: לאיזה קובץ קופל ומה נוסף.\n"
            )
            await conn.execute(
                """INSERT INTO issues
                       (id, company_id, project_id, title, description,
                        status, priority, issue_number, identifier, assignee_agent_id)
                   VALUES ($1, $2::uuid, $3::uuid, $4, $5, 'todo', 'low',
                           $6, $7, $8::uuid)""",
                issue_id, company_id, project_id,
                f"[לקח] קיפול הערת יו\"ר: {short}"[:200],
                description,
                issue_number, identifier, ceo_id,
            )
        finally:
            await conn.close()
    except Exception as e:
        logger.exception("wake_ceo_for_feedback_fold: DB step failed: %s", e)
        return {"ok": False, "error": f"db: {e}"}

    payload = {
        "source": "automation",
        "triggerDetail": "feedback_resolved",
        "reason": f"feedback_fold_{identifier}",
        "payload": {
            "issueId": issue_id,
            "mutation": "feedback_fold",
            "feedbackId": feedback_id,
        },
    }
    try:
        resp = await pc_request(
            "POST", f"/api/agents/{ceo_id}/wakeup",
            json=payload, raise_on_error=True,
        )
        logger.info("feedback-fold wakeup queued: issue=%s feedback=%s", identifier, feedback_id)
        return {"ok": True, "issue_id": issue_id, "identifier": identifier, "wakeup": resp.json()}
    except Exception as e:
        logger.exception("wake_ceo_for_feedback_fold: wakeup API failed: %s", e)
        return {"ok": False, "error": f"wakeup: {e}", "issue_id": issue_id, "identifier": identifier}


async def wake_ceo_agent(issue_id: str, case_number: str, company_id: str = "") -> dict:
    """Wake the CEO agent via Paperclip's wakeup API.

    MUST use API, never direct DB insert (agent won't wake from DB insert).
    Routes to the correct CEO based on company_id.
    """
    if not PAPERCLIP_BOARD_API_KEY:
        raise RuntimeError("PAPERCLIP_BOARD_API_KEY not set — cannot wake CEO agent")

    ceo_id = CEO_AGENTS.get(company_id, CEO_AGENT_ID)
    payload = {
        "source": "on_demand",
        "triggerDetail": "manual",
        "reason": f"start_workflow_{case_number}",
        "payload": {
            "issueId": issue_id,
            "mutation": "workflow_start",
        },
    }

    resp = await pc_request(
        "POST",
        f"/api/agents/{ceo_id}/wakeup",
        json=payload,
        raise_on_error=True,
    )
    result = resp.json()
    logger.info("CEO agent wakeup for case %s: %s", case_number, result)
    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 to run a deterministic generation action (TaskMaster #214).

    Drives the UI "הפקת מסמכים" buttons:
      action="party_claims_summary" → summarize_party_claims (שלב H2)
      action="interim_draft"        → write_interim_draft + export_interim_draft (שלב H)

    A **child issue assigned to the CEO** is created under the case's main issue
    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``.

    Wakeup goes through the Paperclip API (``POST /api/agents/{id}/wakeup``), never
    a direct DB insert (CLAUDE.md hard rule). Returns
    ``{"status": "ok", "issue_id"(=child), "ceo_id", ...}`` or
    ``{"status": "skipped", "reason": ...}``.
    """
    if not PAPERCLIP_BOARD_API_KEY:
        logger.warning("PAPERCLIP_BOARD_API_KEY not set — skipping CEO action wakeup")
        return {"status": "skipped", "reason": "no_api_key"}

    issues = await get_case_issues(case_number)
    if not issues:
        logger.warning("No Paperclip issues found for case %s — skipping CEO action", case_number)
        return {"status": "skipped", "reason": "no_issue"}

    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",
        json={
            "source": "on_demand",
            "triggerDetail": "manual",
            "reason": f"generate_{action}_{case_number}",
            "payload": {
                "issueId": sub_issue_id,
                "action": action,
                "case_number": case_number,
            },
        },
        raise_on_error=True,
    )
    logger.info(
        "CEO action wakeup for case %s: action=%s child_issue=%s ceo=%s status=%s",
        case_number, action, sub_issue_id, ceo_id, wake_resp.status_code,
    )
    return {
        "status": "ok",
        "action": action,
        "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.

    task='learning' — draft↔final voice distillation + the 2-judge style panel.
    task='halacha'  — extract the halachot CITED in the final + corroboration + the
                      3-judge halacha panel.
    The curator (Hermes) runs ONE deterministic local pipeline script per task — the
    script chains the MCP-tool calls + panels internally, so the agent only has to run a
    single command (reliable). Panels write only reversible, CSV-backed proposals (INV-G10).
    """
    if task == "halacha":
        title = f"[ערר {case_number}] אימות-הלכות — פאנל 3-סוכנים"
        description = (
            f"אימות ההלכות סביב ההחלטה הסופית של תיק {case_number} "
            f"(`{final_filename}`).\n\n"
            f"**הרץ פקודה אחת:**\n"
            f"```\ncd /home/chaim/legal-ai/mcp-server && "
            f".venv/bin/python ../scripts/final_halacha_pipeline.py --case {case_number}\n```\n"
            f"הסקריפט מבצע דטרמיניסטית: (1) `extract_internal_citations` (גרף-ציטוטים), "
            f"(2) `corroboration_rebuild` (אות-תיקוף + מדיניות), (3) פאנל-הלכות תלת-סוכני "
            f"(Opus+DeepSeek+Gemini) `--apply` — הסכמה→approved/rejected (הפיך, מגובה ל-CSV), "
            f"פיצול→pending_review ליו\"ר. סמכות binding/persuasive נגזרת מ-precedent_level "
            f"(INV-DM7), לא נקבעת בפאנל.\n"
            f"**לסיום:** כתוב comment בעברית עם סיכום הפלט (אושרו/נדחו/הוסלמו), סגור issue (done). "
            f"תקדים מצוטט שחסר בספרייה — פתח `missing_precedent_create` והפנה ליו\"ר."
        )
        return title, description

    # default: learning (voice distillation + style panel)
    title = f"[ערר {case_number}] סקירת ידע — Knowledge Curator"
    description = (
        f"דפנה סימנה את ההחלטה הסופית של תיק {case_number} כסופית.\n"
        f"קובץ סופי: `{final_filename}`\n\n"
        f"**הרץ פקודה אחת:**\n"
        f"```\ncd /home/chaim/legal-ai/mcp-server && "
        f".venv/bin/python ../scripts/final_learning_pipeline.py --case {case_number}\n```\n"
        f"הסקריפט מבצע דטרמיניסטית: (1) `ingest_final_version` — דיסטילציית טיוטה↔סופי "
        f"(Opus), מסווג style_method מול substance (INV-LRN5), שומר ב-draft_final_pairs "
        f"(status→analyzed); (2) רישום לקורפוס-הסגנון (idempotent); (3) פאנל-סגנון דו-סוכני "
        f"(DeepSeek+Gemini) `--apply` — הסכמה 2/2→decision_lesson "
        f"(source=panel:deepseek+gemini), פיצול→ליו\"ר. substance מדולג.\n"
        f"**לסיום:** מתוך לקחי-הסגנון שאושרו, בחר 3-5 דפוסים שלא תועדו ב-skills/decision/"
        f"SKILL.md / docs/legal-decision-lessons.md / daphna-voice-fingerprint.md, כתוב "
        f"comment בעברית ניטרלי וממוספר, עדכן MEMORY.md, וסגור issue (done). הטמעה "
        f"ל-SKILL.md/lessons.md נשארת אישור-יו\"ר ידני (INV-G10)."
    )
    return title, description


async def wake_curator_for_final(
    case_number: str,
    final_filename: str,
    company_id: str = "",
    task: str = "learning",
) -> dict:
    """Wake the Knowledge Curator (Hermes) for a staged final-decision task.

    Creates a child issue under the main case issue, assigns it to the
    curator, and triggers wakeup. Best-effort — silently skips if no
    curator is configured for the company or no main issue is found.

    ``task`` selects the brief: 'learning' (voice distillation + style panel) or
    'halacha' (cited-halacha extraction + corroboration + halacha panel).

    Returns ``{"status": "ok"|"skipped", ...}``.
    """
    if not PAPERCLIP_BOARD_API_KEY:
        logger.warning("PAPERCLIP_BOARD_API_KEY not set — skipping curator wakeup")
        return {"status": "skipped", "reason": "no_api_key"}

    curator_id = CURATOR_AGENTS.get(company_id)
    if not curator_id:
        logger.info("No curator configured for company %s — skipping", company_id)
        return {"status": "skipped", "reason": "no_curator", "company_id": company_id}

    issues = await get_case_issues(case_number)
    if not issues:
        logger.warning("No Paperclip issues found for case %s — skipping curator", 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"]

    title, description = _curator_task_brief(task, case_number, final_filename)
    child_resp = await pc_request(
        "POST",
        f"/api/issues/{main_issue_id}/children",
        json={
            "title": title,
            "description": description,
            "status": "in_progress",
            "priority": "low",
            "assigneeAgentId": curator_id,
        },
        raise_on_error=True,
    )
    sub_issue = child_resp.json()
    sub_issue_id = sub_issue["id"]

    # Tag plugin_state for case-number visibility on the case page
    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)

    # Trigger wakeup (use API per Paperclip rule — never DB insert)
    wake_resp = await pc_request(
        "POST",
        f"/api/agents/{curator_id}/wakeup",
        json={
            "source": "on_demand",
            "triggerDetail": "manual",
            "reason": f"final_{task}_{case_number}",
            "payload": {
                "issueId": sub_issue_id,
                "mutation": "assignment",
            },
        },
        raise_on_error=True,
    )
    logger.info(
        "Curator wakeup for case %s: sub_issue=%s curator=%s wake=%s",
        case_number, sub_issue_id, curator_id, wake_resp.status_code,
    )
    return {
        "status": "ok",
        "sub_issue_id": sub_issue_id,
        "curator_id": curator_id,
        "main_issue_id": main_issue_id,
    }


async def wake_analyst_for_appraiser_facts(
    case_number: str,
    company_id: str,
) -> dict:
    """Wake the legal-analyst to extract appraiser facts for this case.

    Triggered by the chair clicking "חלץ עובדות שמאיות עכשיו" in the UI.
    The FastAPI container cannot run `extract_appraiser_facts` directly —
    the extractor calls `claude_session.query_json()`, which only works
    where the local `claude` CLI is present (the MCP server / agent runner
    on the host). So instead of running it inline, we create a child issue
    under the case's main Paperclip issue, assign it to the analyst of the
    correct company, and trigger a wakeup with `mutation: extract_appraiser_facts`.
    The analyst's HEARTBEAT picks up the issue, runs the MCP tool locally,
    and reports back via a comment.

    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 %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"]

    description = (
        f"חיים תייג שומות בתיק {case_number} וביקש חילוץ עובדות שמאיות.\n\n"
        f"הרץ `mcp__legal-ai__extract_appraiser_facts(case_number=\"{case_number}\")` "
        f"וכתוב comment בעברית עם תוצאת החילוץ — מספר תכניות, מספר היתרים, "
        f"וסתירות (אם יש) בין שמאים. אם המסמכים חסרי תיוג `appraiser_side`, "
        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",
            # Paperclip ISSUE_PRIORITIES = critical|high|medium|low — "normal"
            # is NOT a valid enum value and the createChildIssueSchema (Zod)
            # rejects it with 400, which surfaces to the chair as a 500.
            "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"extract_appraiser_facts_{case_number}",
            # Use "assignment" — the same mutation `wake_curator_for_final`
            # sends. The HEARTBEAT recognises it; the task-specific intent
            # is conveyed by the child-issue's description, not the payload.
            "payload": {
                "issueId": sub_issue_id,
                "mutation": "assignment",
                "caseNumber": case_number,
            },
        },
        raise_on_error=True,
    )
    logger.info(
        "Analyst wakeup for case %s: sub_issue=%s analyst=%s wake=%s",
        case_number, sub_issue_id, analyst_id, wake_resp.status_code,
    )
    return {
        "status": "queued",
        "sub_issue_id": sub_issue_id,
        "analyst_id": analyst_id,
        "main_issue_id": main_issue_id,
    }


async def wake_analyst_for_argument_aggregation(
    case_number: str,
    company_id: str,
    force: bool = False,
) -> dict:
    """Wake the legal-analyst to aggregate raw claims into legal arguments.

    Triggered by the chair clicking "חשב טיעונים" in the case page. The
    FastAPI container cannot run `aggregate_claims_to_arguments` directly —
    the aggregator calls `claude_session.query_json()`, which only works
    where the local `claude` CLI is present (the MCP server / agent runner
    on the host), **not in this container**. So instead of an in-container
    BackgroundTask (which fails silently and, on `force`, destructively
    deletes the existing arguments before the doomed LLM call), we create a
    child issue under the case's main Paperclip issue, assign it to the
    analyst of the correct company, and trigger a wakeup. The analyst's
    HEARTBEAT picks up the issue, runs the MCP tool locally, and reports
    back via a comment.

    Mirrors ``wake_analyst_for_appraiser_facts`` — same delegation shape.

    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 argument aggregation 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"]

    force_clause = ", force=True" if force else ""
    rerun_note = (
        "זהו חישוב-מחדש (force) — הטיעונים הקיימים יימחקו ויחושבו מחדש.\n\n"
        if force else ""
    )
    description = (
        f"חיים ביקש חישוב טיעונים משפטיים בתיק {case_number}.\n\n"
        f"{rerun_note}"
        f"הרץ `mcp__legal-ai__aggregate_claims_to_arguments(case_number=\"{case_number}\"{force_clause})` "
        f"וכתוב comment בעברית עם תוצאת האיגוד — כמה טיעונים מובחנים נוצרו לכל צד "
        f"(עוררים / משיבים / ועדה / מבקשי-היתר). אם אין טענות גולמיות בתיק, דווח "
        f"ב-comment שצריך להריץ קודם חילוץ טענות (`extract_claims`) וסגור את ה-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",
            # Paperclip ISSUE_PRIORITIES = critical|high|medium|low — "normal"
            # is NOT a valid enum value (Zod 400 → surfaces as 500).
            "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"aggregate_arguments_{case_number}",
            # "assignment" is the generic mutation the HEARTBEAT recognises;
            # task intent lives in the child-issue description, not the payload.
            "payload": {
                "issueId": sub_issue_id,
                "mutation": "assignment",
                "caseNumber": case_number,
            },
        },
        raise_on_error=True,
    )
    logger.info(
        "Analyst wakeup for argument aggregation on case %s: sub_issue=%s "
        "analyst=%s force=%s wake=%s",
        case_number, sub_issue_id, analyst_id, force, wake_resp.status_code,
    )
    return {
        "status": "queued",
        "sub_issue_id": sub_issue_id,
        "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,
    }