Phase 6: preserve provider observation identity

src/poller/forwarder.py

@@ -17,17 +17,44 @@ def github_delivery_id(event_type: str, *identity: object) -> str:
     return f"poller-{event_type}-{digest}"
 
 
-def jira_delivery_id() -> str:
-    """Build a unique delivery ID for one synthetic Jira event."""
-    return f"poller-jira-{uuid4()}"
+def jira_delivery_id(payload: dict[str, Any] | None = None) -> str:
+    """Build a replay-stable ID when a Jira provider identity is available.
+
+    The random fallback is retained for legacy callers that do not provide a
+    webhook-shaped payload.  Provider comment IDs and issue ``updated`` values
+    are stable across poller restarts and therefore make the forwarded
+    delivery interchangeable with a native Jira webhook at Forge.
+    """
+    if payload is None:
+        return f"poller-jira-{uuid4()}"
+    issue = payload.get("issue", {})
+    key = issue.get("key", "") if isinstance(issue, dict) else ""
+    comment = payload.get("comment", {})
+    if isinstance(comment, dict) and comment.get("id") is not None:
+        identity = ("comment", key, comment["id"])
+    else:
+        fields = issue.get("fields", {}) if isinstance(issue, dict) else {}
+        updated = fields.get("updated") if isinstance(fields, dict) else None
+        if isinstance(updated, str) and updated:
+            identity = ("issue", key, updated)
+        else:
+            changelog = payload.get("changelog", {})
+            items = changelog.get("items") if isinstance(changelog, dict) else None
+            if items:
+                identity = ("changelog", key, items)
+            else:
+                return f"poller-jira-{uuid4()}"
+    raw_identity = "\x1f".join(str(part) for part in identity)
+    digest = hashlib.sha256(raw_identity.encode()).hexdigest()[:24]
+    return f"poller-jira-{digest}"
 
 
 async def forward_jira(
     payload: dict[str, Any], delivery_id: str | None = None
 ) -> None:
     settings = get_settings()
     url = f"{settings.forge_gateway_url}/api/v1/webhooks/jira"
-    delivery_id = delivery_id or jira_delivery_id()
+    delivery_id = delivery_id or jira_delivery_id(payload)
     headers = {
         "Content-Type": "application/json",
         "X-Atlassian-Webhook-Identifier": delivery_id,

src/poller/jira.py

@@ -24,7 +24,7 @@ def __init__(self) -> None:
         }
 
     async def get_issue(self, key: str) -> dict[str, Any]:
-        fields = "summary,issuetype,status,labels,comment"
+        fields = "summary,issuetype,status,labels,comment,updated"
         async with httpx.AsyncClient(headers=self._headers) as client:
             r = await client.get(
                 f"{self._base}/rest/api/3/issue/{key}",

src/poller/models.py

@@ -26,6 +26,10 @@ class TicketState:
     summary: str
     labels: set[str]
     last_comment_id: str | None
+    # Native Jira issue revision from fields.updated.  This is delivery
+    # metadata for generating equivalent webhook-shaped observations, not a
+    # workflow checkpoint or poll cursor.
+    issue_updated: str | None = None
     prs: list[PrState] = field(default_factory=list)
     # PRD proposals PR tracking
     prd_pr_repo: str | None = None

src/poller/payloads.py

@@ -8,17 +8,23 @@ def label_changed(
     summary: str,
     old_labels: set[str],
     new_labels: set[str],
+    updated: str | None = None,
 ) -> dict[str, Any]:
+    fields: dict[str, Any] = {
+        "issuetype": {"name": issue_type},
+        "status": {"name": status},
+        "summary": summary,
+        "labels": sorted(new_labels),
+    }
+    if updated:
+        # Jira's issue updated timestamp is the native revision available to
+        # the REST poller for label/status changes.
+        fields["updated"] = updated
     return {
         "webhookEvent": "jira:issue_updated",
         "issue": {
             "key": ticket_key,
-            "fields": {
-                "issuetype": {"name": issue_type},
-                "status": {"name": status},
-                "summary": summary,
-                "labels": sorted(new_labels),
-            },
+            "fields": fields,
         },
         "changelog": {
             "items": [
@@ -43,26 +49,37 @@ def comment_created(
     author_account_id: str,
     author_display_name: str,
     author_email: str = "",
+    comment_id: str | int | None = None,
+    issue_updated: str | None = None,
+    comment_created: str | None = None,
 ) -> dict[str, Any]:
+    fields: dict[str, Any] = {
+        "issuetype": {"name": issue_type},
+        "status": {"name": status},
+        "summary": summary,
+        "labels": sorted(labels),
+    }
+    if issue_updated:
+        fields["updated"] = issue_updated
+    comment: dict[str, Any] = {
+        "body": body,
+        "author": {
+            "accountId": author_account_id,
+            "displayName": author_display_name,
+            "emailAddress": author_email,
+        },
+    }
+    if comment_id is not None:
+        comment["id"] = comment_id
+    if comment_created:
+        comment["created"] = comment_created
     return {
         "webhookEvent": "comment_created",
         "issue": {
             "key": ticket_key,
-            "fields": {
-                "issuetype": {"name": issue_type},
-                "status": {"name": status},
-                "summary": summary,
-                "labels": sorted(labels),
-            },
-        },
-        "comment": {
-            "body": body,
-            "author": {
-                "accountId": author_account_id,
-                "displayName": author_display_name,
-                "emailAddress": author_email,
-            },
+            "fields": fields,
         },
+        "comment": comment,
         "user": {"accountId": author_account_id, "displayName": author_display_name},
     }
 
@@ -96,19 +113,32 @@ def pr_review_submitted(
     review_state: str,
     review_body: str,
     reviewer_login: str,
+    review_id: object | None = None,
+    head_sha: str = "",
+    base_branch: str = "",
+    pr_body: str = "",
+    pr_state: str = "open",
+    draft: bool = False,
 ) -> dict[str, Any]:
+    review: dict[str, Any] = {
+        "state": review_state.lower(),
+        "body": review_body,
+        "user": {"login": reviewer_login},
+    }
+    if review_id is not None:
+        review["id"] = review_id
     return {
         "action": "submitted",
-        "review": {
-            "state": review_state.lower(),
-            "body": review_body,
-        },
+        "review": review,
         "pull_request": {
             "number": pr_number,
             "title": pr_title,
-            "state": "open",
+            "body": pr_body,
+            "state": pr_state,
             "html_url": pr_url,
-            "head": {"ref": branch},
+            "head": {"ref": branch, "sha": head_sha},
+            "base": {"ref": base_branch},
+            "draft": draft,
         },
         "repository": {"full_name": repo},
         "sender": {"login": reviewer_login},
@@ -120,13 +150,28 @@ def issue_comment(
     pr_number: int,
     comment_body: str,
     sender_login: str,
+    comment_id: object | None = None,
+    issue_title: str = "",
+    issue_body: str = "",
+    issue_state: str = "open",
+    issue_url: str = "",
 ) -> dict[str, Any]:
+    comment: dict[str, Any] = {"body": comment_body}
+    if comment_id is not None:
+        comment["id"] = comment_id
+    issue: dict[str, Any] = {
+        "number": pr_number,
+        "title": issue_title,
+        "body": issue_body,
+        "state": issue_state,
+        "html_url": issue_url,
+    }
+    if issue_url:
+        issue["pull_request"] = {"html_url": issue_url}
     return {
         "action": "created",
-        "issue": {"number": pr_number},
-        "comment": {
-            "body": comment_body,
-        },
+        "issue": issue,
+        "comment": comment,
         "repository": {"full_name": repo},
         "sender": {"login": sender_login},
     }
@@ -138,16 +183,21 @@ def pr_merged(
     pr_number: int,
     pr_title: str,
     pr_url: str,
+    head_sha: str = "",
+    pr_body: str = "",
+    base_branch: str = "",
 ) -> dict[str, Any]:
     return {
         "action": "closed",
         "pull_request": {
             "number": pr_number,
             "merged": True,
             "title": pr_title,
+            "body": pr_body,
             "state": "closed",
             "html_url": pr_url,
-            "head": {"ref": branch},
+            "head": {"ref": branch, "sha": head_sha},
+            "base": {"ref": base_branch},
         },
         "repository": {"full_name": repo},
         "sender": {"login": "poller"},