fix(artifacts): record initial publication receipts atomically

src/opensquilla/artifact_session/repository.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import secrets
 import time
@@ -568,6 +569,7 @@ async def _append_audit(
         lease_id: str | None = None,
         payload: dict[str, Any] | None = None,
         created_at: int | None = None,
+        event_id: str | None = None,
     ) -> None:
         await conn.execute(
             """
@@ -578,7 +580,7 @@ async def _append_audit(
             ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
-                self._id_factory("audit"),
+                event_id if event_id is not None else self._id_factory("audit"),
                 document_id,
                 event_type,
                 actor.kind.value,
@@ -827,6 +829,7 @@ async def adopt_generated_deliverable(
         deliverable: ArtifactBlobRef,
         actor: Actor,
         working_source: dict[str, str] | None = None,
+        publication_id: str = "",
     ) -> tuple[CommitResult, DocumentSourceBinding, bool]:
         """Atomically adopt and bind one public generated deliverable.
 
@@ -980,6 +983,20 @@ async def adopt_generated_deliverable(
                     conn, commit=commit, session_key=session_key, session_id=session_id,
                     working_source=working_source,
                 )
+                if publication_id and not rows:
+                    # The initial snapshot is already this occurrence's version.
+                    # Record its receipt before another publisher can advance the
+                    # head, so delayed adoption cannot reapply the old snapshot.
+                    identity = f"{commit.document.document_id}\0{publication_id}".encode()
+                    await self._append_audit(
+                        conn,
+                        document_id=commit.document.document_id,
+                        event_type="document.source_published",
+                        event_id=f"working-publish:{hashlib.sha256(identity).hexdigest()}",
+                        actor=actor,
+                        revision_id=commit.revision.revision_id,
+                        payload={"artifact_id": deliverable.artifact_id},
+                    )
             return commit, binding, True
 
     async def retire_legacy_html_state(self) -> None:

src/opensquilla/artifact_session/service.py

@@ -265,6 +265,7 @@ async def adopt_generated_deliverable(
         deliverable: ArtifactBlobRef,
         actor: Actor,
         working_source: dict[str, str] | None = None,
+        publication_id: str = "",
     ) -> tuple[CommitResult, DocumentSourceBinding, bool]:
         """Adopt one immutable generated deliverable as a stable Document.
 
@@ -282,6 +283,10 @@ async def adopt_generated_deliverable(
             deliverable=_blob(deliverable),
             actor=_actor(actor),
             working_source=working_source,
+            publication_id=(
+                _bounded_text(publication_id, "publication_id", max_bytes=512)
+                if publication_id else ""
+            ),
         )
 
     async def reserve_document_import_attempt(

src/opensquilla/gateway/generated_artifact_adoption.py

@@ -319,6 +319,7 @@ async def __call__(self, event: ArtifactEvent) -> None:
             session_id=self.session_id,
             ref=ref,
             working_source=source,
+            publication_id=event.publication_id,
         )
         if adopted is None:
             return

src/opensquilla/gateway/workbench_resource_runtime.py

@@ -557,6 +557,7 @@ async def adopt_generated_deliverable_if_editable(
     ref: ArtifactRef,
     actor: Actor | None = None,
     working_source: dict[str, str] | None = None,
+    publication_id: str = "",
 ) -> tuple[Document, Revision, DocumentSourceBinding, bool] | None:
     """Associate a generated HTML deliverable and its complete resource bundle with a document.
 
@@ -615,6 +616,7 @@ async def adopt_generated_deliverable_if_editable(
         ),
         actor=actor or Actor(kind=ActorKind.SYSTEM, actor_id="generated-deliverable"),
         working_source=working_source,
+        publication_id=publication_id,
     )
     return commit.document, commit.revision, binding, created
 

tests/functional/test_gateway_attachment_history_e2e.py

@@ -1297,6 +1297,12 @@ async def test_gateway_free_text_attachment_id_does_not_replay_archived_image(
     from opensquilla.session.attachment_manifest import build_attachment_manifest
 
     monkeypatch.setattr(squilla_router_step, "_get_strategy", lambda _cfg: _TextTierStrategy())
+    # Random attachment tokens can contain complaint terms such as "sb" and
+    # upgrade the text route, independently of archived-image replay.
+    monkeypatch.setattr(
+        "opensquilla.gateway.transcripts._new_attachment_id",
+        lambda: "att_abcdefghijklmnopQRstuvwx",
+    )
     manager: SessionManager = _e2e_stack["manager"]
     runner: TurnRunner = _e2e_stack["runner"]
     sink: _EventSink = _e2e_stack["sink"]

tests/test_gateway/test_generated_source_continuity.py

@@ -472,6 +472,57 @@ async def test_concurrent_first_source_publications_create_one_document(
         assert (await cursor.fetchone())[0] == 1
 
 
+@pytest.mark.parametrize("changed", [False, True])
+async def test_initial_source_receipt_survives_a_concurrent_later_publication(
+    source_environment, monkeypatch: pytest.MonkeyPatch, changed: bool,
+):
+    import opensquilla.gateway.generated_artifact_adoption as adoption
+
+    env = source_environment
+    adopters, events = [], []
+    for index in range(2):
+        context, adopter = _turn(env.service, env.store, env.workspace, env.media, env.preview)
+        if index and changed:
+            env.style.write_text("h1{color:crimson}")
+        token = current_tool_context.set(context)
+        try:
+            await publish_artifact(path="index.html")
+        finally:
+            current_tool_context.reset(token)
+        adopters.append(adopter)
+        events.append(ArtifactEvent(**context.published_artifacts[0]))
+
+    created, release = asyncio.Event(), asyncio.Event()
+    original = adoption.adopt_generated_deliverable_if_editable
+
+    async def pause_after_creation(**kwargs):
+        result = await original(**kwargs)
+        if result is not None and result[3]:
+            created.set()
+            await release.wait()
+        return result
+
+    monkeypatch.setattr(adoption, "adopt_generated_deliverable_if_editable", pause_after_creation)
+    first = asyncio.create_task(adopters[0](events[0]))
+    try:
+        await asyncio.wait_for(created.wait(), timeout=5)
+        await adopters[1](events[1])
+    finally:
+        release.set()
+        await first
+
+    binding = await _binding(env)
+    revisions = await env.service.list_revisions(binding.document_id)
+    assert len(revisions) == (2 if changed else 1)
+    head = await env.service.get_document_head(binding.document_id)
+    assert head.revision.artifact_id == events[1 if changed else 0].id
+    # Replaying either completed occurrence must not revert the newer version.
+    await adopters[0](events[0])
+    await adopters[1](events[1])
+    assert await env.service.get_document_head(binding.document_id) == head
+    assert await env.service.list_revisions(binding.document_id) == revisions
+
+
 async def test_source_identity_is_scoped_to_session_and_workspace(source_environment):
     env = source_environment
     await _publish(env.context, env.adopter)