Add wait_for_agent_session for coding-agent follow-up

README.md

@@ -111,6 +111,29 @@ if result.get("isError"):
 
 Calls under `client.fixes` never retry. Results retain untrusted-text wrappers, availability, assessment gaps, and `isError`. `PENDING` and `NOT_VERIFIED` do not establish a fix, and a tool request never records a human disposition.
 
+## Coding-agent sessions
+
+`wait_for_agent_session` polls `get_agent_session` until Proofline stops following the session (`done=True`) or the deadline expires, using the same bounded exponential backoff as `wait_for_review` (5 seconds, doubling to 60 seconds; default 300 seconds).
+
+```python
+from proofline import StartSession
+
+prompt = client.get_agent_session_prompt(org_id, finding_id)
+if prompt.status == "READY":
+    started = client.start_agent_session(
+        org_id,
+        finding_id,
+        body=StartSession(
+            connection_id=prompt.connections[0].id,
+            prompt_digest=prompt.prompt_digest,
+        ),
+    )
+    outcome = client.wait_for_agent_session(org_id, started.id, timeout=300)
+    print(outcome.status, outcome.reason)
+```
+
+`AgentSessionOutcome` retains the latest `AgentSession` on `outcome.session` (excluded from `repr`) and reports the recorded terminal status (`FINISHED`, `FAILED`, `REFUSED`, `HEAD_MOVED`, `UNKNOWN`) or `TIMED_OUT`. `FINISHED` indicates that Proofline stopped following the vendor session; it does not establish that a finding was fixed. Treat `untrusted_vendor_summary` and any vendor-opened pull request URL as untrusted data.
+
 ## Development
 
 The repository bundles its public API contract in `openapi.public.json`. New public operations generate typed methods on both `ProoflineClient` and `AsyncProoflineClient` along with the code in `proofline/_generated/`.

proofline/__init__.py

@@ -33,10 +33,18 @@
 from ._transport import ProoflineError
 from .client import AsyncProoflineClient, ProoflineClient, ReviewFilters
 from .exports import ExportDataset, ExportFormat, ExportPage
-from .outcomes import AssessmentGap, CheckGap, ReviewOutcome
+from .outcomes import (
+    AgentSessionOutcome,
+    AgentSessionStatus,
+    AssessmentGap,
+    CheckGap,
+    ReviewOutcome,
+)
 
 __all__ = [
     "AgentSession",
+    "AgentSessionOutcome",
+    "AgentSessionStatus",
     "AgentVendor",
     "AssessmentGap",
     "AsyncProoflineClient",

proofline/client.py

@@ -73,7 +73,7 @@
 from .exports import ExportDataset, ExportFormat, ExportPage, decode_export
 from .fixes import AsyncFixesClient, FixesClient
 from .knowledge import AsyncKnowledgeClient, KnowledgeClient
-from .outcomes import ReviewOutcome, outcome
+from .outcomes import AgentSessionOutcome, ReviewOutcome, agent_session_outcome, outcome
 
 T = TypeVar("T")
 AgentSessionPrompt = AgentSessionPromptType0 | AgentSessionPromptType1
@@ -506,6 +506,39 @@ def wait_for_review(
         finally:
             _transport.DEADLINE.reset(deadline)
 
+    def wait_for_agent_session(
+        self, organization_id: UUID, session_id: UUID, *, timeout: float = 300
+    ) -> AgentSessionOutcome:
+        """Wait until Proofline finishes following an agent session or return TIMED_OUT."""
+
+        if not math.isfinite(timeout) or timeout < 0:
+            raise ValueError(
+                "The agent session timeout must be a finite nonnegative number of seconds."
+            )
+        last: AgentSession | None = None
+        deadline = _transport.DEADLINE.set(monotonic() + timeout)
+        interval = 5.0
+        try:
+            while (_transport.remaining() or 0) > 0:
+                try:
+                    last = self.get_agent_session(organization_id, session_id)
+                except ProoflineError as error:
+                    if error.status == 0 and (_transport.remaining() or 0) <= 0:
+                        break
+                    raise
+                if (_transport.remaining() or 0) <= 0:
+                    break
+                result = agent_session_outcome(last, session_id=session_id)
+                if result is not None:
+                    return result
+                sleep(min(interval, _transport.remaining() or 0))
+                interval = min(60, interval * 2)
+            return agent_session_outcome(
+                last, "TIMED_OUT", session_id=session_id
+            ) or AgentSessionOutcome("TIMED_OUT", reason="No agent session result is available.")
+        finally:
+            _transport.DEADLINE.reset(deadline)
+
 
 class AsyncProoflineClient(AsyncOperations):
     def __init__(
@@ -852,3 +885,34 @@ async def wait_for_review(
             return outcome(last, url, "TIMED_OUT") or ReviewOutcome("TIMED_OUT", url)
         finally:
             _transport.DEADLINE.reset(deadline)
+
+    async def wait_for_agent_session(
+        self, organization_id: UUID, session_id: UUID, *, timeout: float = 300
+    ) -> AgentSessionOutcome:
+        if not math.isfinite(timeout) or timeout < 0:
+            raise ValueError(
+                "The agent session timeout must be a finite nonnegative number of seconds."
+            )
+        last: AgentSession | None = None
+        deadline = _transport.DEADLINE.set(monotonic() + timeout)
+        interval = 5.0
+        try:
+            while (_transport.remaining() or 0) > 0:
+                try:
+                    last = await self.get_agent_session(organization_id, session_id)
+                except ProoflineError as error:
+                    if error.status == 0 and (_transport.remaining() or 0) <= 0:
+                        break
+                    raise
+                if (_transport.remaining() or 0) <= 0:
+                    break
+                result = agent_session_outcome(last, session_id=session_id)
+                if result is not None:
+                    return result
+                await async_sleep(min(interval, _transport.remaining() or 0))
+                interval = min(60, interval * 2)
+            return agent_session_outcome(
+                last, "TIMED_OUT", session_id=session_id
+            ) or AgentSessionOutcome("TIMED_OUT", reason="No agent session result is available.")
+        finally:
+            _transport.DEADLINE.reset(deadline)

proofline/outcomes.py

@@ -3,17 +3,22 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Literal
+from typing import Literal, cast
 from uuid import UUID
 
+from ._generated.models.agent_session import AgentSession
+from ._generated.models.agent_session_target import AgentSessionTarget
 from ._generated.models.customer_code_assessment import CustomerCodeAssessment
 from ._generated.models.customer_review_detail import CustomerReviewDetail
 from ._generated.types import Unset
 from ._transport import ProoflineError
 
 OutcomeStatus = Literal["COMPLETED", "INCOMPLETE", "FAILED", "STOPPED", "TIMED_OUT"]
+AgentSessionStatus = Literal["FINISHED", "FAILED", "REFUSED", "HEAD_MOVED", "UNKNOWN", "TIMED_OUT"]
 COMPLETE_CHECKS = frozenset(("READY", "BLOCKED", "NOT_APPLICABLE"))
 COMPLETE_LIFECYCLES = frozenset(("GATING", "SHADOW", "QUARANTINED", "RETIRED"))
+ACTIVE_SESSION_STATUSES = frozenset(("REQUESTED", "STARTING", "RUNNING"))
+TERMINAL_SESSION_STATUSES = frozenset(("FINISHED", "FAILED", "REFUSED", "HEAD_MOVED", "UNKNOWN"))
 
 
 @dataclass(frozen=True)
@@ -44,6 +49,13 @@ class ReviewOutcome:
     reason: str | None = None
 
 
+@dataclass(frozen=True)
+class AgentSessionOutcome:
+    status: AgentSessionStatus
+    session: AgentSession | None = field(default=None, repr=False)
+    reason: str | None = None
+
+
 def text(value: object) -> str | None:
     return value if isinstance(value, str) and value else None
 
@@ -56,6 +68,14 @@ def _invalid() -> ProoflineError:
     )
 
 
+def _invalid_agent_session() -> ProoflineError:
+    return ProoflineError(
+        0,
+        "urn:proofline:sdk:invalid-response",
+        "The agent session has invalid identity or follow state.",
+    )
+
+
 def _optional_text(value: object) -> bool:
     return value is None or isinstance(value, (str, Unset))
 
@@ -192,3 +212,67 @@ def outcome(
         if not gaps:
             gaps.append(AssessmentGap(None, review.status, ("REVIEW_STATE_UNAVAILABLE",)))
     return ReviewOutcome(status, review_url, review, tuple(gaps))
+
+
+def _validate_agent_session(session: AgentSession, session_id: UUID | None) -> None:
+    target = session.target
+    if (
+        not isinstance(session.id, UUID)
+        or (session_id is not None and session.id != session_id)
+        or not isinstance(session.finding_id, UUID)
+        or not isinstance(session.review_id, UUID)
+        or not isinstance(session.connection_id, UUID)
+        or not isinstance(session.requested_by_user_id, UUID)
+        or type(session.done) is not bool
+        or type(session.attempt) is not int
+        or session.attempt < 1
+        or not text(session.status)
+        or not isinstance(target, AgentSessionTarget)
+        or not isinstance(target.repository_id, UUID)
+        or not text(target.repository_full_name)
+        or not text(target.head_sha)
+        or type(target.github_repository_id) is not int
+        or target.github_repository_id < 1
+        or type(target.pull_request_number) is not int
+        or target.pull_request_number < 1
+        or not _optional_text(session.cause)
+        or not _optional_text(session.observed_head_sha)
+        or not _optional_text(session.pull_request_url)
+        or not _optional_text(session.untrusted_vendor_summary)
+        or not _optional_text(session.vendor_status)
+        or not _optional_text(session.vendor_url)
+    ):
+        raise _invalid_agent_session()
+
+
+def agent_session_outcome(
+    session: AgentSession | None,
+    forced: AgentSessionStatus | None = None,
+    *,
+    session_id: UUID | None = None,
+) -> AgentSessionOutcome | None:
+    if session is None:
+        return AgentSessionOutcome(
+            forced or "UNKNOWN", reason="No agent session result is available."
+        )
+    _validate_agent_session(session, session_id)
+    if forced is not None:
+        return AgentSessionOutcome(
+            forced,
+            session,
+            reason=(
+                "The agent session did not finish before the timeout."
+                if forced == "TIMED_OUT"
+                else None
+            ),
+        )
+    if not session.done:
+        return None
+    if session.status in TERMINAL_SESSION_STATUSES:
+        return AgentSessionOutcome(cast(AgentSessionStatus, session.status), session)
+    reason = (
+        "The agent session ended without a terminal status."
+        if session.status in ACTIVE_SESSION_STATUSES
+        else "The agent session ended with an unrecognized status."
+    )
+    return AgentSessionOutcome("UNKNOWN", session, reason)

scripts/contracts.py

@@ -31,7 +31,7 @@ def main() -> None:
     from proofline import AsyncProoflineClient, ProoflineClient
 
     for client in (ProoflineClient, AsyncProoflineClient):
-        helpers = {"close", "aclose", "iter_reviews", "wait_for_review"}
+        helpers = {"close", "aclose", "iter_reviews", "wait_for_agent_session", "wait_for_review"}
         exposed = {
             name
             for name in dir(client)

tests/fixtures.py

@@ -96,3 +96,32 @@ def page(*, cursor: str | None = None, truncated: bool = False) -> dict[str, Any
         "truncated": truncated,
         "next_cursor": cursor,
     }
+
+
+def agent_session(status: str = "RUNNING", *, done: bool = False) -> dict[str, Any]:
+    return {
+        "id": str(REVIEW),
+        "finding_id": str(FINDING),
+        "review_id": str(ASSESSMENT),
+        "connection_id": str(REPOSITORY),
+        "vendor": "CURSOR_CLOUD_AGENTS",
+        "target": {
+            "repository_id": str(REPOSITORY),
+            "repository_full_name": "repo/name",
+            "github_repository_id": 42,
+            "pull_request_number": 7,
+            "head_sha": "a" * 40,
+        },
+        "attempt": 1,
+        "status": status,
+        "done": done,
+        "cause": None,
+        "vendor_status": None,
+        "vendor_url": "https://vendor.example/session",
+        "pull_request_url": None,
+        "observed_head_sha": None,
+        "untrusted_vendor_summary": None,
+        "requested_by_user_id": str(ORG),
+        "created_at": "2026-01-01T00:00:00Z",
+        "updated_at": "2026-01-01T00:00:00Z",
+    }