diff --git a/src/tin_lite/activities.py b/src/tin_lite/activities.py index a373408e..f62456a1 100644 --- a/src/tin_lite/activities.py +++ b/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, diff --git a/src/tin_lite/workflow_prerequisites.py b/src/tin_lite/workflow_prerequisites.py index f4036e64..f82b4608 100644 --- a/src/tin_lite/workflow_prerequisites.py +++ b/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 " diff --git a/tests/test_email_outreach.py b/tests/test_email_outreach.py index 58f28542..3d02f339 100644 --- a/tests/test_email_outreach.py +++ b/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 diff --git a/tests/test_workflow_prerequisites.py b/tests/test_workflow_prerequisites.py index 981b343b..58dbe907 100644 --- a/tests/test_workflow_prerequisites.py +++ b/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