From 5a4ab2e7f2d54afa8959ecd113bd7f58cd7e6742 Mon Sep 17 00:00:00 2001 From: egeozin Date: Tue, 13 Oct 2026 00:23:43 -0700 Subject: [PATCH] Read an email campaign's shortlist at the revision admission pinned The campaign looked for outreach/email/SHORTLIST.csv by scanning only the last four commits on the project branch. Once four unrelated commits had landed, every campaign failed with "project state has no email shortlist revision". Admission already pins the revision where it found the shortlist in the run's prerequisite evidence. The campaign now reads the shortlist there, a retry keeps the revision an earlier attempt recorded, and only runs admitted before that evidence existed fall back to the old lookup. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/tin_lite/activities.py | 29 ++++-- src/tin_lite/workflow_prerequisites.py | 18 ++++ tests/test_email_outreach.py | 125 +++++++++++++++++++++++++ tests/test_workflow_prerequisites.py | 16 ++++ 4 files changed, 179 insertions(+), 9 deletions(-) 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