feat(seance): predecessor by-case לצריכת-סוכן דרך ה-plugin (#220 המשך)

מוסיף get_predecessor_for_case(case_number) — האח case-scoped של
get_predecessor_context (by-issue). סוכנים חושבים ב-case_number (לא issue-UUID),
לכן זהו הנתיב שכלי-ה-plugin legal_predecessor_context (PR נפרד ב-plugin-legal-ai)
יקרא. מחזיר מסקנות (summary) של ריצות-heartbeat שהסתיימו על-פני issues של התיק,
newest-first, עם summary בלבד (ריצות cancelled ריקות מסוננות).

- shaper משותף _shape_predecessor_run (G2 — נתיב-shaping יחיד ל-by-issue+by-case).
- join בטוח i.id::text = payload->>'issueId' (לא cast של payload חופשי ל-uuid).
- Port pc_get_predecessor_for_case (read-only); endpoint
  GET /api/operations/cases/{case_number}/predecessor.
- 2 בדיקות (shaping+identifier, case-לא-נמצא). אומת חי: 1043-02-26/8125-09-24
  מחזירים מסקנות-דיספוזיציה אמיתיות.

Invariants: G12 (fetch מהמעטפת, המגע מהשער; הכלי יחיה ב-plugin=מעטפת מוצהרת, לא
ב-mcp-server), G2 (shaper יחיד). המשך: PR ב-plugin-legal-ai (כלי) + הוראת-HEARTBEAT.

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

web/agent_platform_port.py

@@ -43,6 +43,7 @@ from web.paperclip_client import (
     get_agent_health as pc_get_agent_health,
     get_recent_escalations as pc_get_recent_escalations,
     get_predecessor_context as pc_get_predecessor_context,
+    get_predecessor_for_case as pc_get_predecessor_for_case,
     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,
@@ -183,6 +184,7 @@ __all__ = [
     "pc_get_agent_health",
     "pc_get_recent_escalations",
     "pc_get_predecessor_context",
+    "pc_get_predecessor_for_case",
     "pc_list_live_runs",
     "pc_get_run_log",
     "pc_get_run_events",

web/app.py

@@ -65,6 +65,7 @@ from web.agent_platform_port import (
     pc_get_agent_health,
     pc_get_recent_escalations,
     pc_get_predecessor_context,
+    pc_get_predecessor_for_case,
     pc_get_agents,
     pc_get_agents_for_case,
     pc_get_case_issues,
@@ -7615,6 +7616,17 @@ async def operations_issue_predecessor(issue_id: str, limit: int = 3):
     return await pc_get_predecessor_context(issue_id, limit)
 
 
+@app.get("/api/operations/cases/{case_number}/predecessor")
+async def operations_case_predecessor(case_number: str, limit: int = 5):
+    """Recent finished runs across a case's issues (#220, agent-facing "seance").
+
+    Case-scoped sibling of the by-issue endpoint — the plugin tool
+    ``legal_predecessor_context`` calls this so a resuming agent reads what prior
+    sessions on the case concluded (agents key on case_number, not issue UUID).
+    """
+    return await pc_get_predecessor_for_case(case_number, 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

@@ -1216,22 +1216,68 @@ async def get_predecessor_context(issue_id: str, limit: int = 3) -> dict:
     finally:
         await conn.close()
 
-    runs = [
+    runs = [_shape_predecessor_run(r) for r in rows]
-        {
-            "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}
 
 
+def _shape_predecessor_run(r) -> dict:
+    """Shared row → predecessor-run dict (by-issue and by-case entry points, G2)."""
+    return {
+        "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"],
+    }
+
+
+async def get_predecessor_for_case(case_number: str, limit: int = 5) -> dict:
+    """Recent finished runs across a CASE's issues, newest-first (#220, agent-facing).
+
+    The case-scoped sibling of :func:`get_predecessor_context`. The agent knows its
+    ``case_number`` (not the Paperclip issue UUID), so the plugin tool
+    ``legal_predecessor_context`` calls this. Resolves the case's Paperclip project,
+    then returns finished heartbeat-run summaries across its issues — so a resuming
+    wake reads what prior sessions on the case concluded instead of re-deriving.
+
+    Read-only. The issue join casts the known-valid ``issues.id`` to text (rather
+    than the free-form ``payload->>'issueId'`` to uuid) so a malformed payload
+    can't break the query. ``limit`` clamped 1..10.
+    """
+    limit = max(1, min(int(limit), 10))
+    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": True, "case_number": case_number, "runs": []}
+        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,
+                      i.identifier
+                 FROM heartbeat_runs h
+                 JOIN agent_wakeup_requests w ON w.id = h.wakeup_request_id
+                 JOIN issues i ON i.id::text = w.payload->>'issueId'
+                 LEFT JOIN agents a ON a.id = h.agent_id
+                WHERE i.project_id = $1
+                  AND h.finished_at IS NOT NULL
+                  AND h.result_json->>'summary' IS NOT NULL
+                ORDER BY h.started_at DESC
+                LIMIT $2""",
+            project["id"], limit,
+        )
+    finally:
+        await conn.close()
+
+    runs = [{**_shape_predecessor_run(r), "identifier": r["identifier"]} for r in rows]
+    return {"ok": True, "case_number": case_number, "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 = "ספריית פסיקה — תור חילוץ"

web/tests/test_agent_health.py

@@ -198,3 +198,54 @@ def test_get_recent_escalations_defaults_severity_when_untagged(monkeypatch):
     (item,) = asyncio.run(pc.get_recent_escalations())["items"]
     assert item["severity"] == "medium"  # fallback
     assert item["created_at"] is None
+
+
+# ── predecessor by case (#220 agent-facing) — resolve project then shape runs ──
+class _ProjectRunsConn:
+    def __init__(self, project, runs):
+        self._project, self._runs = project, runs
+
+    async def fetchrow(self, query, *args):
+        return self._project
+
+    async def fetch(self, query, *args):
+        return self._runs
+
+    async def close(self):
+        pass
+
+
+def test_get_predecessor_for_case_shapes_runs(monkeypatch):
+    from datetime import datetime
+
+    runs = [
+        {
+            "id": "run-1", "status": "succeeded",
+            "started_at": datetime(2026, 7, 5, 10, 0, 0),
+            "finished_at": datetime(2026, 7, 5, 10, 5, 0),
+            "summary": "הלולאה נסגרה כראוי", "error_code": None,
+            "session_id_after": "sess-9", "agent_name": "עוזר משפטי",
+            "identifier": "CMP-141",
+        },
+    ]
+
+    async def _connect(_url):
+        return _ProjectRunsConn({"id": "proj-1"}, runs)
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+    result = asyncio.run(pc.get_predecessor_for_case("1043-02-26"))
+    assert result["ok"] is True and result["case_number"] == "1043-02-26"
+    (run,) = result["runs"]
+    assert run["run_id"] == "run-1"
+    assert run["identifier"] == "CMP-141"  # by-case adds which issue
+    assert run["started_at"] == "2026-07-05T10:00:00"
+    assert "הלולאה" in run["summary"]
+
+
+def test_get_predecessor_for_case_unknown_case_is_empty(monkeypatch):
+    async def _connect(_url):
+        return _ProjectRunsConn(None, [])  # no project row
+
+    monkeypatch.setattr(pc.asyncpg, "connect", _connect)
+    result = asyncio.run(pc.get_predecessor_for_case("9999-99-99"))
+    assert result == {"ok": True, "case_number": "9999-99-99", "runs": []}