Read an email campaign's shortlist at the revision admission pinned

src/tin_lite/activities.py

@@ -158,6 +158,7 @@
     PrerequisiteError,
     can_wait_for_prerequisites,
     evaluate_prerequisites,
+    pinned_artifact_revision,
     prerequisite_wait_run_ids,
     waited_evidence,
 )
@@ -2560,15 +2561,25 @@ async def prepare_email_campaign(self, run_id_text: str) -> None:
                     raise RuntimeError(
                         "Google Workspace does not grant the email campaign capabilities"
                     )
-                shortlist_history = await self._storage.canonical_file_history(
-                    repo_id=project.state_repo_id,
-                    branch=project.canonical_branch,
-                    path=EMAIL_SHORTLIST_PATH,
-                    limit=1,
-                )
-                if not shortlist_history:
-                    raise RuntimeError("project state has no email shortlist revision")
-                shortlist_commit_sha = shortlist_history[0]["revision"]
+                # Read the shortlist exactly as admission found it, however many commits have
+                # landed since. A retry keeps the revision an earlier attempt already recorded.
+                campaign = await self._db.get_email_campaign(run_id)
+                shortlist_commit_sha = (
+                    campaign["source_commit_sha"]
+                    if campaign is not None
+                    else pinned_artifact_revision(run.prerequisite_evidence, EMAIL_SHORTLIST_PATH)
+                )
+                if shortlist_commit_sha is None:
+                    # Runs admitted before admission pinned the shortlist look it up as before.
+                    shortlist_history = await self._storage.canonical_file_history(
+                        repo_id=project.state_repo_id,
+                        branch=project.canonical_branch,
+                        path=EMAIL_SHORTLIST_PATH,
+                        limit=1,
+                    )
+                    if not shortlist_history:
+                        raise RuntimeError("project state has no email shortlist revision")
+                    shortlist_commit_sha = shortlist_history[0]["revision"]
                 shortlist = await self._storage.read_canonical_artifact(
                     repo_id=project.state_repo_id,
                     commit_sha=shortlist_commit_sha,

src/tin_lite/workflow_prerequisites.py

@@ -537,6 +537,24 @@ def prerequisite_wait_run_ids(evidence: Any) -> list[str]:
     return list(dict.fromkeys(ids))  # A run and its identity can wait on the same run.
 
 
+def pinned_artifact_revision(evidence: Any, path: str) -> str | None:
+    """The project revision at which admission (or a settled wait) found the file at `path`.
+
+    None when the run pinned no such evidence, as runs admitted before artifact evidence did.
+    """
+    items = evidence.get("items") if isinstance(evidence, Mapping) else None
+    for item in items if isinstance(items, list) else []:
+        if not isinstance(item, Mapping) or not item.get("satisfied"):
+            continue
+        found = item.get("evidence")
+        if not isinstance(found, Mapping) or found.get("path") != path:
+            continue
+        revision = found.get("revision")
+        if isinstance(revision, str) and re.fullmatch(r"[0-9a-f]{40}", revision):
+            return revision
+    return None
+
+
 def _wait_hint(result: PrerequisiteResult) -> str:
     return (
         f"Tin waits up to {PREREQUISITE_WAIT_MINUTES} minutes for "

tests/test_email_outreach.py

@@ -672,3 +672,128 @@ async def test_the_changed_campaign_queries_run_against_the_migrated_schema(publ
     await database.skip_outreach_delivery(execution_key="missing", reason="campaign_stopped")
     with pytest.raises(RuntimeError, match="cannot complete"):
         await database.complete_email_campaign(run_id=UUID(int=12))
+
+
+@pytest.mark.parametrize("recorded", [None, "e" * 40])
+async def test_campaign_reads_the_shortlist_admission_pinned_however_many_commits_ago(
+    recorded,
+) -> None:
+    """Regression: the shortlist was looked up only in the last four commits."""
+    from tin_lite.activities import TinActivities
+    from tin_lite.domain import EMAIL_SHORTLIST_PATH, RunStatus
+
+    pinned = "b" * 40
+    expected = recorded or pinned
+    project = SimpleNamespace(
+        id=UUID("00000000-0000-4000-8000-0000000000dd"),
+        state_repo_id="projects/test",
+        canonical_branch="main",
+    )
+    run = SimpleNamespace(
+        id=RUN_ID,
+        executor=EMAIL_CAMPAIGN_WORKFLOW_NAME,
+        status=RunStatus.RUNNING,
+        project_id=project.id,
+        input={"subject": "Hello {{name}}", "body": "Hi {{name}},\n\nWould Tuesday work?"},
+        prerequisite_evidence={
+            "head_commit_sha": pinned,
+            "items": [
+                {
+                    "kind": "artifact",
+                    "path": EMAIL_SHORTLIST_PATH,
+                    "satisfied": True,
+                    "evidence": {"path": EMAIL_SHORTLIST_PATH, "revision": pinned},
+                }
+            ],
+        },
+    )
+    shortlist = shortlist_csv(
+        (
+            "person-1",
+            "ada@example.com",
+            "Ada Lovelace",
+            "Analytical Engines",
+            "former collaborator",
+            "2026-08-01",
+            "Relevant launch",
+            "selected",
+            "",
+        )
+    )
+
+    class FakeDatabase:
+        created = None
+
+        @asynccontextmanager
+        async def effect_lock(self, execution_key, operation):
+            yield object(), None
+
+        @asynccontextmanager
+        async def project_state_lock(self, conn, value):
+            yield
+
+        async def start_effect(self, conn, **values):
+            return None
+
+        async def complete_effect(self, conn, **values):
+            return None
+
+        async def fail_effect(self, conn, **values):
+            return None
+
+        async def get_run(self, value):
+            return run
+
+        async def get_project(self, value):
+            return project
+
+        async def mark_run_running(self, value):
+            return None
+
+        async def get_integration_connection(self, **values):
+            return SimpleNamespace(
+                id=UUID("00000000-0000-4000-8000-0000000000ee"),
+                status="connected",
+                external_account_id="google-account",
+                external_account_label="founder@example.com",
+                configuration={
+                    "granted_capabilities": [
+                        "gmail.messages.send",
+                        "gmail.messages.read",
+                        "gmail.history.read",
+                    ]
+                },
+            )
+
+        async def get_email_campaign(self, value):
+            return None if recorded is None else {"source_commit_sha": recorded}
+
+        async def create_email_campaign(self, **values):
+            self.created = values
+
+        async def add_activity(self, **values):
+            return None
+
+    class FakeStorage:
+        async def canonical_file_history(self, **values):
+            raise AssertionError("an admitted campaign never scans recent commits")
+
+        async def read_canonical_artifact(self, *, repo_id, commit_sha, path):
+            assert (commit_sha, path) == (expected, EMAIL_SHORTLIST_PATH)
+            return shortlist
+
+        async def publish_state_documents(self, **values):
+            return "c" * 40, True
+
+    database = FakeDatabase()
+    activities = TinActivities(
+        database=database,  # type: ignore[arg-type]
+        storage=FakeStorage(),  # type: ignore[arg-type]
+        sandboxes=SimpleNamespace(),  # type: ignore[arg-type]
+        settings=SimpleNamespace(),  # type: ignore[arg-type]
+    )
+
+    await activities.prepare_email_campaign(str(RUN_ID))
+
+    assert database.created["shortlist_commit_sha"] == expected
+    assert len(database.created["recipients"]) == 1

tests/test_workflow_prerequisites.py

@@ -28,6 +28,7 @@
     PrerequisiteError,
     WorkflowPrerequisite,
     parse_workflow_prerequisites,
+    pinned_artifact_revision,
     project_readiness,
     validate_prerequisite_graph,
 )
@@ -886,3 +887,18 @@ async def test_short_listing_names_only_what_an_agent_needs_to_choose(publicatio
     descriptions = [row["description"] for row in short_rows]
     assert all("\n" not in text and len(text) <= 160 for text in descriptions)
     assert len(json.dumps(short_rows)) * 5 < len(json.dumps(full_rows))
+
+
+def test_pinned_artifact_revision_reads_only_a_satisfied_pin_for_that_path():
+    found = {"path": "outreach/email/SHORTLIST.csv", "revision": HEAD}
+    evidence = {
+        "items": [{"kind": "run", "satisfied": True}, {"satisfied": True, "evidence": found}]
+    }
+    assert pinned_artifact_revision(evidence, "outreach/email/SHORTLIST.csv") == HEAD
+    assert pinned_artifact_revision(evidence, "outreach/email/OTHER.csv") is None
+    unmet = {"items": [{"satisfied": False, "evidence": found}]}
+    assert pinned_artifact_revision(unmet, "outreach/email/SHORTLIST.csv") is None
+    malformed = {"items": [{"satisfied": True, "evidence": {**found, "revision": "HEAD"}}]}
+    assert pinned_artifact_revision(malformed, "outreach/email/SHORTLIST.csv") is None
+    for legacy in (None, {}, {"items": "none"}, {"items": ["not a mapping"]}):
+        assert pinned_artifact_revision(legacy, "outreach/email/SHORTLIST.csv") is None