Resolve native maintenance parent lineage before binding notifications

docs/scm.md

@@ -1071,6 +1071,14 @@ resume, inherited parent grant, extra coding budget, or retry of an uncertain
 remote write. Unbound children, checkpoint tests and other project readers keep
 read-only lineage. Revoked or expired consent holds further notifications.
 
+A parent reached through native checkpoint lineage has two different identities:
+its run-creation receipt proves the derived run, while its authenticated SCM
+parent receipt proves the issue link. Configuration and every later notification
+revalidation resolve both stores and compare the same typed lineage, frozen
+source and current project/component/owner authority. The original creator is
+provenance, not a new execution grant. Newer status comments do not replace the
+retained parent, and no duplicate parent notification is needed.
+
 This first setup path is native API configuration. The natural coordinator does
 not yet create a maintenance unit or its communication binding from an issue
 request. Controlled host/native tests and a real GitHub acceptance trial are

src/firefly_forge/scm_conversations.py

@@ -98,68 +98,100 @@ async def references(self, delivery: Data, scopes: list[Data], *, durable: bool
                 raise AccessDenied("Ticket context exceeds the bounded lookup; review its exact native receipts")
             candidates = [(copy.deepcopy(row.data), row.sequence) for row in rows]
         for previous, sequence in candidates:
-            run_id = previous.get("run_id")
-            if (
-                not run_id
-                or previous["id"] == delivery["id"]
-                or previous.get("natural_parent")
-                or (previous.get("command") or {}).get("adapter")
-            ):
-                continue
             try:
-                run = await self.store.get_run(run_id)
-                scope = next(
-                    (
-                        s
-                        for s in scopes
-                        if s["project_id"] == run.get("project_id") and run.get("component_id") in s["component_ids"]
-                    ),
-                    None,
-                )
-                if scope is None or run.get("repository_id") != delivery["repository_id"]:
-                    continue
-                if previous.get("maintain_ticket") is not None:
-                    from firefly_forge.scm_maintain_tickets import MaintainTickets
-
-                    await MaintainTickets(self.collaboration).authorize_reference(previous, run, delivery, scopes)
-                elif previous.get("native_question") is not None:
-                    from firefly_forge.native_question_publications import NativeQuestionPublications
-
-                    await NativeQuestionPublications(self.collaboration).authorize_reference(previous, run)
-                elif (previous.get("verification") or {}).get("transport") == "github_rest_poll_v1":
-                    polling = self.collaboration.polling or getattr(self.store, "scm_polling", None)
-                    if polling is None:
-                        continue
-                    await polling.ticket_reference(previous, run)
-                else:
-                    await SCMActorAuthority(self.store, self.collaboration.scm).ticket_reference(previous, run)
+                reference, run = await self._receipt_reference(previous, sequence, delivery, scopes)
             except (AccessDenied, KeyError, ValueError, TypeError):
                 continue
+            run_id = run["id"]
             if run_id in runs:
                 continue
             runs[run_id] = run
             if len(runs) > 200:
                 raise AccessDenied("The current ticket has too many distinct run references for one natural choice")
-            references.append(
-                TicketContextReferenceV1(
-                    receipt_id=previous["id"],
-                    sequence=sequence,
-                    run_id=run_id,
-                    project_id=run["project_id"],
-                    component_id=run["component_id"],
-                    context_digest=digest(
-                        {
-                            "observation": observation_cut(previous),
-                            "actor": previous.get("scm_execution_actor"),
-                            "run_id": run_id,
-                            "action_status": previous.get("action_status"),
-                        }
-                    ),
-                ).model_dump(mode="json")
-            )
+            references.append(reference)
         await self._checkpoint_tests(delivery, scopes, references, runs)
         return references, runs
 
+    async def _receipt_reference(
+        self, previous: Data, sequence: int, delivery: Data, scopes: list[Data]
+    ) -> tuple[Data, Data]:
+        run_id = previous.get("run_id")
+        if (
+            not run_id
+            or previous["id"] == delivery["id"]
+            or previous.get("natural_parent")
+            or (previous.get("command") or {}).get("adapter")
+            or any(previous.get(key) != delivery.get(key) for key in ("repository_id", "provider"))
+            or any(
+                (previous.get("issue") or {}).get(key) != (delivery.get("issue") or {}).get(key)
+                for key in ("kind", "number")
+            )
+        ):
+            raise AccessDenied("The retained ticket reference has no exact SCM parent")
+        run = await self.store.get_run(run_id)
+        scope = next(
+            (
+                s
+                for s in scopes
+                if s["project_id"] == run.get("project_id") and run.get("component_id") in s["component_ids"]
+            ),
+            None,
+        )
+        if scope is None or run.get("repository_id") != delivery["repository_id"]:
+            raise AccessDenied("The retained ticket polling authority is unavailable")
+        if previous.get("maintain_ticket") is not None:
+            from firefly_forge.scm_maintain_tickets import MaintainTickets
+
+            await MaintainTickets(self.collaboration).authorize_reference(previous, run, delivery, scopes)
+        elif previous.get("native_question") is not None:
+            from firefly_forge.native_question_publications import NativeQuestionPublications
+
+            await NativeQuestionPublications(self.collaboration).authorize_reference(previous, run)
+        elif (previous.get("verification") or {}).get("transport") == "github_rest_poll_v1":
+            polling = self.collaboration.polling or getattr(self.store, "scm_polling", None)
+            if polling is None:
+                raise AccessDenied("The retained ticket reference is outside current native scope")
+            await polling.ticket_reference(previous, run)
+        else:
+            await SCMActorAuthority(self.store, self.collaboration.scm).ticket_reference(previous, run)
+        reference = TicketContextReferenceV1(
+            receipt_id=previous["id"],
+            sequence=sequence,
+            run_id=run_id,
+            project_id=run["project_id"],
+            component_id=run["component_id"],
+            context_digest=digest(
+                {
+                    "observation": observation_cut(previous),
+                    "actor": previous.get("scm_execution_actor"),
+                    "run_id": run_id,
+                    "action_status": previous.get("action_status"),
+                }
+            ),
+        ).model_dump(mode="json")
+        return reference, run
+
+    async def resolve_reference(self, reference: Data, delivery: Data, scopes: list[Data]) -> Data:
+        """Resolve stable typed lineage through its SCM receipt and native creation stores."""
+        retained = TicketContextReferenceV1.model_validate(reference).model_dump(mode="json")
+        receipt_id = (
+            retained["receipt_id"]
+            if retained.get("kind", "scm_receipt") == "scm_receipt"
+            else retained["parent_receipt_id"]
+        )
+        async with self.store._unit():
+            row = await self.collaboration.inbox.find_by_id(identifier(receipt_id))
+            if row is None:
+                raise AccessDenied("The authenticated SCM parent receipt is unavailable")
+            previous, sequence = copy.deepcopy(row.data), row.sequence
+        parent, run = await self._receipt_reference(previous, sequence, delivery, scopes)
+        references = [parent]
+        if retained.get("kind", "scm_receipt") != "scm_receipt":
+            await self._checkpoint_tests(delivery, scopes, references, {run["id"]: run})
+        if sum(item == retained for item in references) != 1:
+            raise AccessDenied("The native creation or authenticated ticket parent provenance changed")
+        return previous
+
     async def _checkpoint_tests(self, delivery: Data, scopes: list[Data], references: list[Data], runs: Data) -> None:
         from firefly_forge.checkpoint_test_models import frozen_checkpoint_source
         from firefly_forge.run_creation_requests import RunCreationRequests

src/firefly_forge/scm_maintain_tickets.py

@@ -138,7 +138,7 @@ async def configure(self, project: str, repository: str, request: MaintainTicket
         parents = [value for value in references if value["run_id"] == parent_id]
         if len(parents) != 1 or run["component_id"] not in scope["component_ids"]:
             raise AccessDenied("The observed ticket has no exact authenticated checkpoint parent")
-        parent = await self.c.receipt(parents[0]["receipt_id"])
+        parent = await SCMConversations(self.c).resolve_reference(parents[0], source, [scope])
         parent_cut = observation_cut(parent)
         await self.store.admit_credentials({"request": incoming, "source": observation_cut(source)})
         async with self.polling._locked(project, repository, [actor.id.hex, owner["user_id"]]):
@@ -149,7 +149,8 @@ async def configure(self, project: str, repository: str, request: MaintainTicket
             current_receipt = await self._creation(request.creation_request_id)
             current_run = await self.store.get_run(request.run_id)
             creation_scope(current_run, current_receipt)
-            if current_receipt != receipt or observation_cut(await self.c.receipt(parent["id"])) != parent_cut:
+            current_parent = await SCMConversations(self.c).resolve_reference(parents[0], source, [current_scope])
+            if current_receipt != receipt or observation_cut(current_parent) != parent_cut:
                 raise AccessDenied("The retained maintain creation or ticket parent changed")
             row, _ = await self.polling._state(repository)
             state = state_of(row.data)
@@ -329,10 +330,10 @@ async def current(self, data: Data, *, remote: bool = False, dispatch: bool = Fa
                 raise AccessDenied("The maintain question changed from its native feedback")
         if creation["digest"] != value["creation_digest"]:
             raise AccessDenied("The maintain child creation changed")
-        parent = await self.c.receipt(value["parent_reference"]["receipt_id"])
+        scope, _, owner = await self._target(source, project, repository, remote=remote)
+        parent = await SCMConversations(self.c).resolve_reference(value["parent_reference"], source, [scope])
         if observation_cut(parent) != value["parent_observation"]:
             raise AccessDenied("The exact native parent ticket provenance changed")
-        scope, _, owner = await self._target(source, project, repository, remote=remote)
         if (
             scope != value["polling_scope"]
             or owner != value["owner"]

tests/test_scm_maintain_derived_parent_host.py

@@ -0,0 +1,119 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Native creation provenance must resolve to its distinct authenticated SCM parent."""
+
+import copy
+from types import SimpleNamespace
+from uuid import UUID, uuid4
+
+import pytest
+from test_scm_maintain_ticket_host import body, prepared_binding
+from test_scm_polling import poll_case as poll_case
+
+from firefly_forge.journal import digest
+from firefly_forge.project_access import AccessDenied
+from firefly_forge.run_creation_requests import RunCreationRequests
+from firefly_forge.scm_maintain_ticket_models import MaintainTicketInput
+from firefly_forge.scm_maintain_tickets import MaintainTickets
+
+
+async def nested_binding(c, monkeypatch):
+    service, first = await prepared_binding(c, monkeypatch)
+    parent = c.db["runs"][first.run_id].data
+    parent_creation = await service._creation(first.creation_request_id)
+    parent_creation["original_actor"] = {"user_id": c.user_id}
+    parent_creation["digest"] = digest({k: v for k, v in parent_creation.items() if k != "digest"})
+    child_id, creation_id = uuid4().hex, uuid4().hex
+    reference = {**parent["source_checkpoint"], "run_id": parent["id"], "request_id": uuid4().hex}
+    source = {**copy.deepcopy(parent["source_checkpoint_snapshot"]), "reference": reference}
+    source["digest"] = digest({k: v for k, v in source.items() if k != "digest"})
+    child = {
+        **copy.deepcopy(parent),
+        "id": child_id,
+        "source_checkpoint": reference,
+        "source_checkpoint_snapshot": source,
+    }
+    c.db["runs"][child_id] = SimpleNamespace(
+        id=UUID(child_id), repository_id=UUID(c.repository), status="queued", data=child
+    )
+    creation = {
+        **copy.deepcopy(parent_creation),
+        "request_id": creation_id,
+        "run_id": child_id,
+        "source_checkpoint": source,
+        "input": {"operation": "maintain", "publish": False, "source_checkpoint": reference},
+    }
+    creation["digest"] = digest({k: v for k, v in creation.items() if k != "digest"})
+    creations = {parent_creation["request_id"]: parent_creation, creation_id: creation}
+
+    async def load(self, request_id):
+        return copy.deepcopy(creations.get(request_id))
+
+    async def children(parents, repository):
+        assert repository == UUID(c.repository)
+        return [UUID(parent_creation["request_id"])] if UUID(c.run_id) in parents else []
+
+    monkeypatch.setattr(RunCreationRequests, "_load", load)
+    monkeypatch.setattr(MaintainTickets, "_creation", load)
+    c.store.run_creation_records = SimpleNamespace(checkpoint_children=children)
+    request = MaintainTicketInput.model_validate(
+        {
+            **body(),
+            "request_id": uuid4().hex,
+            "run_id": child_id,
+            "creation_request_id": creation_id,
+            "expected_creation_digest": creation["digest"],
+            "observed_receipt_id": first.observed_receipt_id,
+            "expected_reference_digest": first.expected_reference_digest,
+        }
+    )
+    c.state["derived_parent_posts"] = copy.deepcopy(c.state["posts"])
+    return service, request, creations, parent
+
+
+async def test_native_derived_parent_configures_and_revalidates_without_treating_creation_as_inbox(
+    poll_case, monkeypatch
+):
+    async with poll_case() as c:
+        service, request, creations, parent = await nested_binding(c, monkeypatch)
+        before = copy.deepcopy(parent)
+        kept = await service.configure(c.project, c.repository, request)
+        ref = kept["parent_reference"]
+        assert ref["kind"] == "native_checkpoint_maintain"
+        assert ref["receipt_id"] in creations and ref["receipt_id"] not in c.db["inbox"]
+        assert ref["parent_receipt_id"] == kept["parent_observation"]["id"]
+        anchor = await c.collaboration.receipt(kept["outbox_id"])
+        assert (await service.current(anchor))["run"]["id"] == request.run_id
+        assert c.db["runs"][parent["id"]].data == before
+        assert c.state["posts"] == c.state["derived_parent_posts"]
+
+
+@pytest.mark.parametrize("change", ["digest", "creator", "source", "scope"])
+async def test_derived_parent_provenance_drift_refuses_current_notification(poll_case, monkeypatch, change):
+    async with poll_case() as c:
+        service, request, creations, parent = await nested_binding(c, monkeypatch)
+        kept = await service.configure(c.project, c.repository, request)
+        creation = creations[kept["parent_reference"]["receipt_id"]]
+        if change == "digest":
+            creation["digest"] = "f" * 64
+        elif change == "creator":
+            creation["original_actor"]["user_id"] = "f" * 32
+        elif change == "source":
+            c.db["runs"][parent["id"]].data["source_checkpoint_snapshot"]["revision"] = "f" * 40
+        else:
+            creation["component_id"] = "other"
+        anchor = await c.collaboration.receipt(kept["outbox_id"])
+        with pytest.raises((AccessDenied, ValueError)):
+            await service.current(anchor)
+        assert c.state["posts"] == c.state["derived_parent_posts"]
+
+
+async def test_newer_ticket_reference_does_not_replace_the_bound_native_parent(poll_case, monkeypatch):
+    async with poll_case() as c:
+        service, request, _, _ = await nested_binding(c, monkeypatch)
+        kept = await service.configure(c.project, c.repository, request)
+        previous = copy.deepcopy(kept["parent_reference"])
+        await c.new_command("status", number=102)
+        anchor = await c.collaboration.receipt(kept["outbox_id"])
+        context = await service.current(anchor)
+        assert context["binding"]["parent_reference"] == previous
+        assert context["run"]["id"] == request.run_id