web/agent_platform_port.py

@@ -41,6 +41,7 @@ from web.paperclip_client import (
     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,
@@ -179,6 +180,7 @@ __all__ = [
     "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/app.py

@@ -63,6 +63,7 @@ from web.agent_platform_port import (
     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,
@@ -7544,6 +7545,18 @@ async def operations_agent_health():
     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

@@ -1132,6 +1132,54 @@ async def get_agent_health() -> dict:
     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 = "ספריית פסיקה — תור חילוץ"

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": []}