Phase 6: reconcile webhook and poller observations

docs/architecture/internals.md

@@ -16,7 +16,11 @@ Gateway and Worker communicate only through Redis and can be deployed on separat
 
 **Checkpointing:** LangGraph workflow state is persisted via `AsyncRedisSaver`, keyed by Jira ticket key (e.g., `AISOS-123`). Checkpoints are written after each graph node completes. When a new event arrives for an existing ticket, the workflow resumes from its last checkpoint.
 
-**Idempotency:** A `DeduplicationService` exists but is not yet wired into the webhook routes. Branch creation and label operations are naturally idempotent; Jira comment posting is not.
+**Idempotency:** The worker records normalized webhook/poller observations in
+the reconciliation ledger before command interpretation. Duplicate, stale,
+and conflicting observations do not re-enter the workflow. External writes
+still use durable effect identities; provider operations such as Jira comment
+posting must remain idempotent across crash recovery.
 
 **Consistency boundary:** Workflow mutations are persisted as stable effect intents before
 provider execution. Provider-specific recovery evidence and idempotent ref updates close

docs/architecture/option-b-completion-plan.md

@@ -125,7 +125,7 @@ Exit gate: the supported process can be rendered from a versioned definition, an
 
 ## Phase 6 — Reconciliation and poller convergence
 
-PR: new stacked PR based on Phase 5. Status: missing.
+PR: #331 (`phase6/reconciliation-contract`). Status: complete.
 
 Purpose: make webhooks and the existing forge-poller equivalent observation sources while workflow state remains authoritative for transition progress.
 
@@ -141,6 +141,17 @@ Work:
 
 Exit gate: lost, duplicated, and reordered delivery converges to the same workflow/effect state without duplicate effects.
 
+Evidence: the versioned Observation fixtures in
+`tests/contracts/fixtures/observations/` and
+`tests/contracts/fixtures/reconciliation/`, adapter and identity tests in
+`tests/contracts/` and `tests/unit/integrations/source_control/`, ledger
+classification tests in `tests/unit/reconciliation/`, and replay/worker
+convergence tests in `tests/contracts/reconciliation/` and
+`tests/unit/orchestrator/test_reconciliation_worker.py`. The explicit
+no-native-revision limitation is documented in
+`docs/architecture/phase-6-observation-contract.md`; such input is retained
+for operator review and never used to infer workflow position.
+
 ## Phase 7 — Execution read models and operations
 
 PR: #329, rebased onto Phase 6. Status: partial.
@@ -181,7 +192,7 @@ Exit gate: all supported flows use normalized observations, authoritative comman
 2. Finish #326, then rebase all descendants.
 3. Finish #327, then rebase all descendants.
 4. Finish #328.
-5. Create `phase6/reconciliation-contract` from the Phase 5 tip and open its stacked PR.
+5. Phase 6 is implemented by Forge PR #331 and the companion forge-poller PR #10.
 6. Rebase `phase7/execution-read-models` from the old Phase 5 base onto Phase 6; change #329's base to the Phase 6 branch.
 7. Rebase `phase8-compatibility-removal` onto the rebased Phase 7 branch; retain #330's Phase 7 base.
 8. Finish Phase 6, then Phase 7, then Phase 8, rebasing descendants after each phase changes.

docs/architecture/phase-6-observation-contract.md

@@ -0,0 +1,54 @@
+# Phase 6 Observation contract
+
+Forge accepts webhook and `forge-poller` deliveries through the same ingress
+adapters. Both paths are normalized to the versioned `Observation` record
+(`schema_version: "1.0"`). The strict contract contains source,
+provider/resource identity, provider revision, observation times, normalized
+facts, correlation metadata, and an optional evidence reference. Unknown
+fields and future schema versions are rejected at the boundary.
+
+## Identity
+
+`observation_id` identifies the provider event record. It is generated from
+the provider event ID plus provider/resource identity and revision context; it
+does not include `source`. Replays that preserve the provider event ID retain
+the same observation ID. When a poller must use a different transport ID,
+`delivery_identity` still remains stable whenever the provider revision is
+present.
+
+`delivery_identity` is the deduplication identity used by Forge's observation
+ledger. It is generated from source system, resource identity, and
+`resource_revision`. Therefore webhook and poller deliveries of one provider
+revision have the same delivery identity even when their transport event IDs,
+received times, or source values differ. `revision_order` is ordering metadata
+and is not included when a provider revision is available. For resources with
+no revision, the provider event identity is used; callers must provide
+`correlation.provider_event_id` (or `transport_event_id`) in that case.
+
+The Jira adapter derives revisions from the immutable comment ID, the issue's
+`updated` timestamp, or a changelog fingerprint. Source-control adapters use
+the change-request head SHA, check state scoped to its commit, or immutable
+comment/review IDs. The poller can therefore forward its existing Jira and
+GitHub payloads; Forge assigns the source-independent identity before ledger
+processing. Command IDs likewise use `delivery_identity`, so a poller retry
+cannot create a second command/effect for a webhook-delivered revision.
+
+An observation without a native provider revision is deliberately limited: it
+is deduplicated only when its provider event ID is stable, and a later
+revision cannot be ordered safely. Forge records such input as an
+operator-visible conflict rather than guessing which external state is newer.
+This is the explicit no-native-revision limitation, not a workflow checkpoint.
+
+The shared fixtures at
+[`github_pull_request_revision.json`](../../tests/contracts/fixtures/observations/github_pull_request_revision.json)
+and
+[`source_control_sequence.json`](../../tests/contracts/fixtures/reconciliation/source_control_sequence.json)
+show equivalent source-control revisions and replay behavior. Contract tests
+cover both ingress markers, cross-source deduplication, monotonic stale/reorder
+handling, command identity, and the Jira revision derivation cases.
+
+The companion poller changes preserve GitHub review/comment IDs and head SHAs,
+include Jira issue `updated` values and comment IDs/timestamps in forwarded
+payloads, and make synthetic delivery IDs replay-stable. The poller contract
+tests cover those payload guarantees; Forge's adapter tests then verify their
+identity and reconciliation semantics.

docs/architecture/phase-6-reconciliation-contract.md

@@ -0,0 +1,56 @@
+# Phase 6: reconciliation contract
+
+Phase 6 makes webhook and polling delivery interchangeable inputs to Forge.
+Workflow position remains owned by the workflow instance; an external
+observation may update an external-state projection but cannot set
+`current_node`, workflow identity, or transition counters.
+
+## Conformance contract
+
+Every ingress source must provide a versioned `Observation` with:
+
+- `source`: `webhook` or `poller` (transport metadata only);
+- provider/resource identity (`source_system`, `resource`, and
+  `resource_revision`);
+- a stable provider event identity in `correlation.provider_event_id`;
+- monotonic `revision_order` whenever a provider revision cannot be ordered
+  from its native identifier; and
+- provider facts that are identical for equivalent revisions.
+
+`Observation.delivery_identity` intentionally excludes `source`, so equivalent
+webhook and poller deliveries deduplicate. A command may be evaluated only for
+an accepted observation. Duplicate, stale, or conflicting deliveries are
+recorded for inspection and cannot create another external effect.
+
+The shared fixture is
+[`source_control_sequence.json`](../../tests/contracts/fixtures/reconciliation/source_control_sequence.json).
+It is JSON rather than a Python fixture so Forge and `forge-poller` can replay
+the same provider revisions. Forge's contract tests cover equivalent envelopes
+and replay the sequence with a lost first revision, duplicate cross-source
+delivery, stale reordered delivery, and a duplicate replay. The resulting
+latest revision, accepted command, and effect list must match the clean replay.
+
+Run the Forge side with:
+
+```shell
+pytest -q tests/contracts/reconciliation tests/unit/reconciliation
+```
+
+## Poller companion behavior and limitation
+
+`forge-poller` remains an independently deployable delivery source. Its
+webhook-shaped Jira and GitHub payloads preserve the provider IDs and revision
+metadata needed by Forge's normalizing adapters, which assign the
+source-independent revision and delivery identity before workflow evaluation.
+Poller cursors are delivery optimizations only; they are not read or written
+as workflow checkpoints.
+Consequently a lost, repeated, or reordered delivery cannot move workflow
+position or create a duplicate command/effect.
+
+The provider limitation is explicit: if a payload contains neither a native
+revision nor a stable provider event ID, Forge cannot establish ordering. It
+records the observation but classifies a competing update as
+`operator_required`; it does not infer chronology from receipt time or a
+poller cursor. Jira payloads lacking `issue.fields.updated`, changelog items,
+or a comment ID are in this category and require the provider/poller to add a
+stable revision before automatic convergence is possible.

docs/architecture/reference.md

@@ -26,7 +26,9 @@ Workflows pause at defined gates and wait indefinitely for human approval. The `
 
 - **No PEL reclaim**: Unacknowledged messages from crashed workers remain in Redis PEL indefinitely. Recovery requires manual `XCLAIM`.
 - **No distributed per-ticket lock**: Multiple workers can process events for the same ticket concurrently, causing potential checkpoint conflicts.
-- **Webhook deduplication not wired**: `DeduplicationService` exists but is not connected to webhook routes.
+- **Ingress delivery is at-least-once**: webhook and poller observations are
+  deduplicated and classified by the reconciliation ledger, but the gateway
+  may still enqueue a retried transport message before the worker records it.
 - **Webhook signature validation is optional**: Endpoints accept unsigned payloads when secrets are not configured.
 - **No approval gate timeout**: Paused workflows wait indefinitely with no escalation.
 - **Single Redis dependency**: No Sentinel, Cluster, or HA. Redis is a single point of failure.

src/forge/domain/__init__.py

@@ -9,7 +9,12 @@
     stable_identity,
 )
 from forge.domain.interactions import CommentType, classify_comment
-from forge.domain.observations import Observation, ObservationSource
+from forge.domain.observations import (
+    Observation,
+    ObservationSource,
+    observation_delivery_identity,
+    observation_identity,
+)
 from forge.domain.schema import DomainModel, JsonValue, VersionedDomainModel
 from forge.domain.stations import (
     StationFailure,
@@ -27,6 +32,8 @@
     "JsonValue",
     "Observation",
     "ObservationSource",
+    "observation_delivery_identity",
+    "observation_identity",
     "ResourceIdentity",
     "StationFailure",
     "StationInvocationIdentity",

src/forge/domain/observations.py

@@ -7,7 +7,7 @@
 
 from pydantic import Field
 
-from forge.domain.identity import ResourceIdentity
+from forge.domain.identity import ResourceIdentity, stable_identity
 from forge.domain.schema import JsonValue, VersionedDomainModel
 
 
@@ -23,8 +23,76 @@ class Observation(VersionedDomainModel):
     source_system: str = Field(min_length=1)
     resource: ResourceIdentity
     resource_revision: str | None = None
+    revision_order: int | None = Field(default=None, ge=0)
     observed_at: datetime
     received_at: datetime
     facts: dict[str, JsonValue] = Field(default_factory=dict)
     correlation: dict[str, JsonValue] = Field(default_factory=dict)
     evidence_reference: str | None = None
+
+    @property
+    def delivery_identity(self) -> str:
+        """Identity shared by poller and webhook deliveries of the same revision."""
+        return observation_delivery_identity(self)
+
+
+def observation_delivery_identity(observation: Observation) -> str:
+    """Return the source-independent identity of an observation delivery.
+
+    A provider revision is the strongest identity available: delivery IDs are
+    transport metadata and differ when the same state arrives from a webhook
+    and from the poller.  For event-shaped resources that do not expose a
+    revision, the provider event ID (stored in correlation metadata) keeps
+    distinct events from collapsing into one delivery.  ``observation_id`` is
+    the final fallback for callers constructing an observation without either
+    kind of provider identity.
+
+    ``revision_order`` is deliberately not included when ``resource_revision``
+    is present.  It is ordering metadata, not part of the provider revision;
+    including it would make equivalent deliveries deduplicate differently.
+    """
+    parts: dict[str, JsonValue] = {
+        "source_system": observation.source_system,
+        "resource_type": observation.resource.resource_type,
+        "external_id": observation.resource.external_id,
+        "namespace": observation.resource.namespace,
+    }
+    if observation.resource_revision is not None:
+        parts["resource_revision"] = observation.resource_revision
+    elif observation.revision_order is not None:
+        parts["revision_order"] = observation.revision_order
+    else:
+        provider_event_id = observation.correlation.get("provider_event_id")
+        if not isinstance(provider_event_id, str):
+            provider_event_id = observation.correlation.get("transport_event_id")
+        if isinstance(provider_event_id, str) and provider_event_id:
+            parts["provider_event_id"] = provider_event_id
+        else:
+            parts["observation_id"] = observation.observation_id
+    return stable_identity("observation-delivery", parts)
+
+
+def observation_identity(
+    *,
+    source_system: str,
+    provider_event_id: str,
+    resource: ResourceIdentity,
+    resource_revision: str | None = None,
+) -> str:
+    """Build the deterministic identity assigned to a provider observation.
+
+    This identity remains stable when the delivery source changes.  The event
+    ID distinguishes separate provider events, while the resource revision is
+    included as context for providers that reuse event IDs across resources.
+    """
+    return stable_identity(
+        "observation",
+        {
+            "source_system": source_system,
+            "provider_event_id": provider_event_id,
+            "resource_type": resource.resource_type,
+            "external_id": resource.external_id,
+            "namespace": resource.namespace,
+            "resource_revision": resource_revision,
+        },
+    )

src/forge/integrations/source_control/observations.py

@@ -12,7 +12,7 @@
     Observation,
     ObservationSource,
     ResourceIdentity,
-    stable_identity,
+    observation_identity,
 )
 from forge.integrations.source_control.contracts import NormalizedEvent
 
@@ -44,20 +44,25 @@ def normalized_event_to_observation(
     external_id = event.repo_ref.id
     resource_type = "repository"
     revision: str | None = None
-    if change_request:
-        resource_type = "change_request"
-        external_id = f"{event.repo_ref.id}#{native_id}"
-        revision = change_request.head_sha or None
-    elif event.check:
+    # Event-specific resources must win over the optional change-request
+    # context attached by GitHub's check/comment/review payloads.  Otherwise
+    # every check or comment on one head SHA is misidentified as that PR state.
+    if event.check:
         resource_type = "check"
         external_id = f"{event.repo_ref.id}:{event.check.name}"
-        revision = f"{event.check.status.value}:{event.check.conclusion.value}"
+        revision = _check_revision(event)
     elif event.comment:
         resource_type = "comment"
         external_id = f"{event.repo_ref.id}:{event.comment.id}"
+        revision = event.comment.id
     elif event.review:
         resource_type = "review"
         external_id = f"{event.repo_ref.id}:{event.review.id}"
+        revision = event.review.id
+    elif change_request:
+        resource_type = "change_request"
+        external_id = f"{event.repo_ref.id}#{native_id}"
+        revision = change_request.head_sha or None
 
     facts = _json_value(
         {
@@ -72,27 +77,44 @@ def normalized_event_to_observation(
         }
     )
     assert isinstance(facts, dict)
-    observation_id = stable_identity(
-        "observation",
-        {
-            "source_system": event.repo_ref.provider.value,
-            "event_id": event.id,
-            "resource_revision": revision,
-        },
+    resource = ResourceIdentity(
+        resource_type=resource_type,
+        external_id=external_id,
+        namespace=event.repo_ref.connection,
+    )
+    observation_id = observation_identity(
+        source_system=event.repo_ref.provider.value,
+        provider_event_id=event.id,
+        resource=resource,
+        resource_revision=revision,
     )
     return Observation(
         observation_id=observation_id,
         source=source,
         source_system=event.repo_ref.provider.value,
-        resource=ResourceIdentity(
-            resource_type=resource_type,
-            external_id=external_id,
-            namespace=event.repo_ref.connection,
-        ),
+        resource=resource,
         resource_revision=revision,
         observed_at=event.received_at,
         received_at=event.received_at,
         facts=facts,
-        correlation={"transport_event_id": event.id, "repository_id": event.repo_ref.id},
+        correlation={
+            "provider_event_id": event.id,
+            "transport_event_id": event.id,
+            "repository_id": event.repo_ref.id,
+        },
         evidence_reference=f"source-control-event:{event.id}" if event.raw else None,
     )
+
+
+def _check_revision(event: NormalizedEvent) -> str:
+    """Return a revision for a check state, scoped to the checked commit.
+
+    ``CheckRun`` intentionally keeps only provider-neutral fields.  Combining
+    those fields with the associated change-request head prevents a successful
+    check on two commits from being treated as one revision.  For standalone
+    checks, the event ID is the provider's only available immutable identity.
+    """
+    assert event.check is not None
+    head_sha = event.change_request.head_sha if event.change_request else ""
+    scope = head_sha or event.id
+    return f"{scope}:{event.check.status.value}:{event.check.conclusion.value}"

src/forge/orchestrator/event_adapters/commands.py

@@ -141,7 +141,11 @@ def interpret_event(
     command_id = stable_identity(
         "workflow-command",
         {
-            "event_id": message.event_id,
+            # The transport event ID changes when one provider revision is
+            # delivered by both webhook and poller.  Command identity must be
+            # tied to the source-independent observation so either delivery
+            # selects the same durable command/effect.
+            "observation_delivery_identity": adapted.observation.delivery_identity,
             "run_id": workflow.run_id,
             "command_type": command_type.value,
         },